This gist contains the code for parsing job status events from lambda-function.py, which is part of the blog-amazon-macie-custom-eventbridge-events repository.

blog-amazon-macie-custom-eventbridge-eventbridge_lambda-function-parse-job-status-events.py

def parse_job_status_event(log_event):
    """
    Parse a Macie job status event from a CloudWatch Logs log event.

    Macie publishes job status events as JSON objects to CloudWatch Logs.
    The log event message contains the full event details including:
    - eventType (e.g., JOB_CREATED, ONE_TIME_JOB_STARTED, JOB_COMPLETED)
    - jobId
    - jobName
    - occurredAt
    - description
    - And other fields depending on the event type

    Parameters:
        log_event (dict): A single log event from CloudWatch Logs containing
                         'id', 'timestamp', and 'message' fields

    Returns:
        dict: The parsed job status event data, or None if parsing fails
    """
    try:
        message = log_event.get("message", "")

        # Macie log events are JSON formatted
        event_data = json.loads(message)

        # Validate that this is a job status event by checking for required fields
        if "eventType" not in event_data:
            logger.warning("Log event does not contain 'eventType' field")
            return None

        # List of known job status event types
        job_status_event_types = [
            "JOB_CREATED",
            "ONE_TIME_JOB_STARTED",
            "SCHEDULED_RUN_STARTED",
            "SCHEDULED_RUN_COMPLETED",
            "JOB_COMPLETED",
            "JOB_CANCELLED",
            "JOB_PAUSED_BY_USER",
            "JOB_RESUMED_BY_USER",
            "JOB_PAUSED_BY_MACIE_SERVICE_QUOTA_MET",
            "JOB_RESUMED_BY_MACIE_SERVICE_QUOTA_LIFTED",
            "BUCKET_MATCHED_THE_CRITERIA",
            "NO_BUCKETS_MATCHED_THE_CRITERIA"
        ]

        # Only process job status events (not account-level or bucket-level errors)
        if event_data["eventType"] in job_status_event_types:
            # Add metadata to help consumers of the EventBridge event
            event_data["processedAt"] = datetime.now(timezone.utc).isoformat()
            event_data["sourceLogGroup"] = "/aws/macie/classificationjobs"
            event_data["sourceLogEventId"] = log_event.get("id", "")
            return event_data

        logger.info(
            f"Skipping non-job-status event type: {event_data['eventType']}"
        )
        return None

    except json.JSONDecodeError:
        logger.warning(
            f"Log event message is not valid JSON: {log_event.get('message', '')[:200]}"
        )
        return None
    except Exception as e:
        logger.error(f"Error parsing job status event: {str(e)}")
        return None