Dataflow Jobs Hotkey Logging Should Be Enabled

Using Console

To remediate the misconfiguration “Dataflow Jobs Hotkey Logging Should Be Enabled” for GCP using GCP console, you can follow the below steps:

Open the GCP console and navigate to the Dataflow section.
Select the Dataflow job for which you want to enable hotkey logging.
Click on the “Edit” button located on the top of the page.
Scroll down to the “Additional parameters” section.
In the “Additional parameters” section, add the following parameter:

—experiments=enable_hot_key_logging

Click on the “Save” button to save the changes.
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.

Using CLI

To remediate the “Dataflow Jobs Hotkey Logging Should Be Enabled” misconfiguration in GCP using GCP CLI, you can follow these steps:

Open the Cloud Shell in the GCP Console.
Run the following command to enable hotkey logging for Dataflow jobs:

gcloud config set dataflow/logging_severity "INFO"

Run the following command to verify that hotkey logging is enabled:

gcloud config get-value dataflow/logging_severity

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.

Using Python

To remediate the “Dataflow Jobs Hotkey Logging Should Be Enabled” misconfiguration in GCP using Python, you can follow these steps:

Import the necessary libraries:

from googleapiclient import discovery
from oauth2client.client import GoogleCredentials

Set up the credentials:

credentials = GoogleCredentials.get_application_default()
service = discovery.build('logging', 'v2', credentials=credentials)

Define the project ID:

project_id = 'your-project-id'

Define the log name:

log_name = 'dataflow.googleapis.com%2Fjob_message'

Define the filter:

filter_str = 'resource.type="dataflow_step" AND severity>=ERROR AND textPayload:"hotkey"'

Define the sink name:

sink_name = 'hotkey-sink'

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.")

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.")

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.")

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.

Using Terraform

# There is currently no Terraform argument on any google_dataflow_* resource
# that enables Dataflow "hot key logging"; it is a per‑pipeline option set in
# the Dataflow SDK / job invocation, not in GCP infrastructure config.

# In practice, you must:
# - Enable hot key logging in your Dataflow pipeline code (e.g. via the
#   SDK-specific pipeline option such as `--hotKeyLoggingEnabled` or equivalent),
#   or
# - When launching the job/template (Console or gcloud), pass the corresponding
#   runtime flag/parameter that turns on hot key logging.

# Terraform can create/launch jobs (google_dataflow_job, google_dataflow_flex_template_job,
# google_dataflow_streaming_job, etc.), but the provider does NOT expose a
# "hot key logging" field, so this finding cannot be remediated via Terraform
# configuration of the job resource itself.

# Verification from an IaC perspective is not possible with `terraform plan`
# because no diff is produced for this behavior; instead, confirm in:
# - Your Dataflow job code / template options, and
# - The Dataflow job details in the GCP Console that hot key logging is enabled.

GCP Introduction

GCP Pricing

GCP Threats

GCP Misconfigurations

Resources

GCP Threats

Dataflow Jobs Hotkey Logging Should Be Enabled

More Info:

Risk Level

Address

Compliance Standards

Triage and Remediation

Remediation

​More Info:

​Risk Level

​Address

​Compliance Standards

​Triage and Remediation

​Remediation

​