def save_to_gss()

in benchmarks/benchmark/tools/locust-load-inference/locust-runner/app/main.py [0:0]

• 41 lines of code
• 7 McCabe index (conditional complexity)

def save_to_gss(run: LocustRun, stats, metrics_list: [Metric]):
    storage_client = storage.Client()
    bucket_name = os.environ["BUCKET"]

    bucket = storage_client.bucket(bucket_name)
    now = datetime.now()
    timestamp = now.strftime('stats_and_metrics%Y-%m-%d%H:%M:%S')
    blob = bucket.blob(timestamp)

    start_time_formatted = time.strftime(
        "%Y-%m-%d %H:%M:%S", time.gmtime(run.start_time))
    end_time_formatted = time.strftime(
        "%Y-%m-%d %H:%M:%S", time.gmtime(run.end_time))

    with blob.open("w") as f:
        f.write("*********** AI on GKE Benchmarking Tool ******************\n")
        f.write("**********************************************************\n")
        f.write(f"Test duration: {run.duration} [s] \n")
        f.write(f"Test start: {start_time_formatted} \n")
        f.write(f"Test end: {end_time_formatted} \n")
        f.write(f"Users: {run.users} \n")
        f.write(f"Rate: {run.rate} \n")
        f.write("\n**********************************************************\n\n")
        f.write("Locust statistics\n")
        f.write(stats)
        f.write("\n**********************************************************\n\n")
        f.write("\n\nMetrics\n")
        for m in metrics_list:
            f.write(m.name)
            f.write("\n")
            for result in m.results:
                label = result.resource.labels
                # metadata = result.metadata.system_labels.fields
                metricdata = result.metric.labels
                for l in label:
                    f.write(f"""{l}: {label[l]}\n""")
                for l in metricdata:
                    f.write(f"""{l}: {metricdata[l]}\n""")
                points = result.points
                f.write(f"""Number of points: {len(points)}\n""")
                for point in points:
                    p_point = (
                        point.value.double_value) if point.value.double_value is not None else 0
                    f.write(f""" {p_point},""")
                f.write("\n ")