​
More Info:

Dataflow jobs hot key logging should be enabled

​
Risk Level

Low

​
Address

Operational Maturity, Reliability

​
Compliance Standards

CBP

​
Triage and Remediation

​
Remediation

Using Console

To remediate the misconfiguration “Dataflow Jobs Hotkey Logging Should Be Enabled” for GCP using GCP console, you can follow the below steps:
  1. Open the GCP console and navigate to the Dataflow section.
  2. Select the Dataflow job for which you want to enable hotkey logging.
  3. Click on the “Edit” button located on the top of the page.
  4. Scroll down to the “Additional parameters” section.
  5. In the “Additional parameters” section, add the following parameter:
—experiments=enable_hot_key_logging
  1. Click on the “Save” button to save the changes.
  2. Verify that the hotkey logging is enabled by checking the logs for the Dataflow job.
By following the above steps, you can remediate the misconfiguration “Dataflow Jobs Hotkey Logging Should Be Enabled” for GCP using GCP console.

​

To remediate the “Dataflow Jobs Hotkey Logging Should Be Enabled” misconfiguration in GCP using GCP CLI, you can follow these steps:
  1. Open the Cloud Shell in the GCP Console.
  2. Run the following command to enable hotkey logging for Dataflow jobs: 
gcloud config set dataflow/logging_severity "INFO"
  1. Run the following command to verify that hotkey logging is enabled:
gcloud config get-value dataflow/logging_severity
  1. If the output of the above command is “INFO”, then hotkey logging is enabled.
By following these steps, you can remediate the “Dataflow Jobs Hotkey Logging Should Be Enabled” misconfiguration in GCP using GCP CLI.
To remediate the “Dataflow Jobs Hotkey Logging Should Be Enabled” misconfiguration in GCP using Python, you can follow these steps:
  1. Import the necessary libraries:
from googleapiclient import discovery
from oauth2client.client import GoogleCredentials
  1. Set up the credentials:
credentials = GoogleCredentials.get_application_default()
service = discovery.build('logging', 'v2', credentials=credentials)
  1. Define the project ID:
project_id = 'your-project-id'
  1. Define the log name:
log_name = 'dataflow.googleapis.com%2Fjob_message'
  1. Define the filter:
filter_str = 'resource.type="dataflow_step" AND severity>=ERROR AND textPayload:"hotkey"'
  1. Define the sink name:
sink_name = 'hotkey-sink'
  1. Check if the sink exists:
sinks = service.sinks().list(projectId=project_id).execute()
if sink_name in [sink['name'] for sink in sinks['sinks']]:
    print(f"Sink {sink_name} already exists.")
else:
    print(f"Sink {sink_name} does not exist.")
  1. If the sink does not exist, create it:
if sink_name not in [sink['name'] for sink in sinks['sinks']]:
    sink = {
        'name': sink_name,
        'destination': f'bigquery.googleapis.com/projects/{project_id}/datasets/hotkey_sink',
        'filter': filter_str,
        'outputVersionFormat': 'V2'
    }
    service.sinks().create(projectId=project_id, body=sink).execute()
    print(f"Sink {sink_name} created.")
  1. If the sink exists, update it:
else:
    sink_uri = f"projects/{project_id}/sinks/{sink_name}"
    sink = service.sinks().get(sinkName=sink_uri).execute()
    if sink['filter'] == filter_str:
        print(f"Sink {sink_name} already has the correct filter.")
    else:
        sink['filter'] = filter_str
        service.sinks().update(sinkName=sink_uri, body=sink).execute()
        print(f"Sink {sink_name} updated.")
  1. Verify that the sink has been created or updated:
sinks = service.sinks().list(projectId=project_id).execute()
if sink_name in [sink['name'] for sink in sinks['sinks']]:
    sink_uri = f"projects/{project_id}/sinks/{sink_name}"
    sink = service.sinks().get(sinkName=sink_uri).execute()
    print(f"Sink {sink_name} has filter: {sink['filter']}")
else:
    print(f"Sink {sink_name} not found.")
These steps will create or update a sink in GCP that logs hotkeys for Dataflow jobs.