()
| 45 | run_every=timedelta(minutes=5), |
| 46 | ) |
| 47 | def update_sse_usage(): # type: ignore[no-untyped-def] |
| 48 | agg_request_count: dict[str, int] = {} |
| 49 | agg_last_event_generated_at: dict[str, str] = {} |
| 50 | |
| 51 | for log in sse_service.stream_access_logs(): |
| 52 | agg_request_count[log.api_key] = agg_request_count.get(log.api_key, 0) + 1 |
| 53 | agg_last_event_generated_at[log.api_key] = log.generated_at |
| 54 | |
| 55 | with InfluxDBWrapper.get_client().write_api( |
| 56 | write_options=WriteOptions(batch_size=100, flush_interval=1000) |
| 57 | ) as write_api: |
| 58 | environments = Environment.objects.filter( |
| 59 | api_key__in=agg_request_count.keys() |
| 60 | ).values( |
| 61 | "api_key", |
| 62 | "id", |
| 63 | "project_id", |
| 64 | "project__name", |
| 65 | "project__organisation_id", |
| 66 | "project__organisation__name", |
| 67 | ) |
| 68 | |
| 69 | for environment in environments: |
| 70 | time = agg_last_event_generated_at[environment["api_key"]] |
| 71 | count = agg_request_count[environment["api_key"]] |
| 72 | record = ( |
| 73 | Point("sse_call") |
| 74 | .field("request_count", count) |
| 75 | .tag("environment_id", environment["id"]) |
| 76 | .tag("project_id", environment["project_id"]) |
| 77 | .tag("project", environment["project__name"]) |
| 78 | .tag("organisation_id", environment["project__organisation_id"]) |
| 79 | .tag("organisation", environment["project__organisation__name"]) |
| 80 | .time(time) |
| 81 | ) |
| 82 | |
| 83 | write_api.write(bucket=settings.INFLUXDB_BUCKET, record=record) |
| 84 | |
| 85 | |
| 86 | def get_auth_header(): # type: ignore[no-untyped-def] |
searching dependent graphs…