def embedded_metrics_placed_jobs()

in src/BatchJobsStates/JobStatesToCW/app.py [0:0]

• 19 lines of code
• 4 McCabe index (conditional complexity)

def embedded_metrics_placed_jobs(dimensions, properties, metric_name, metric_time, metrics):

    try:
        metrics.set_namespace("AWSBatchMetrics")
        logger.info(f"Deactivate metrics.should_use_default_dimensions {metrics.context.should_use_default_dimensions}")
        metrics.context.should_use_default_dimensions = False
        logger.info(f" Done metrics.should_use_default_dimensions {metrics.context.should_use_default_dimensions}")

        # use the event timestamp instead of using the EMF context timestamp
        metric_converted_time = datetime.datetime.strptime(metric_time, '%Y-%m-%dT%H:%M:%SZ')
        metric_converted_time = int(round(metric_converted_time.timestamp()*1000))
        metrics.context.meta['Timestamp'] = metric_converted_time

        for k, v in dimensions.items():
            metrics.put_dimensions({k:v})
        metrics.put_metric(metric_name, 1, "Count")
        for k, v in properties.items():
            metrics.set_property(k, v)

    except ClientError as e:
        message = f"Error adding CloudWatch metric job resubmission: {format(e)}"
        logger.warning(message)
        raise Exception(message)
    return