puppet: Add vector pipelines for other Akamai SQS queues.

puppet/zulip_ops/manifests/prometheus/akamai.pp

@@ -7,7 +7,10 @@ class zulip_ops::prometheus::akamai {
 
   $bin = $zulip_ops::vector::bin
   $conf = '/etc/vector.toml'
-  $sqs_url = zulipsecret('secrets', 'akamai_sqs_url', '')
+  $pipelines = {
+    'static' => zulipsecret('secrets', 'akamai_static_sqs_url', ''),
+    'realm'  => zulipsecret('secrets', 'akamai_realm_sqs_url', ''),
+  }
 
   file { $conf:
     ensure  => file,

puppet/zulip_ops/templates/vector.toml.template.erb

@@ -1,7 +1,6 @@
 [sources.vector_metrics]
   type = "internal_metrics"
 
-[sources.s3_akamai_static]
 # Akamai Datastream2 logs all accesses into AWS S3:
 # https://techdocs.akamai.com/datastream2/docs/stream-amazon-s3
 #
@@ -10,56 +9,61 @@
 # deletes the messages from the queue, and the S3 bucket is
 # configured to purge old logs.
 # https://vector.dev/docs/reference/configuration/sources/aws_s3/
+
+<% @pipelines.each do |key,sqs_url| %>
+[sources.s3_akamai_<%= key %>]
   type = "aws_s3"
   region = "us-east-1"
   compression = "gzip"
   sqs.delete_message = true
   sqs.poll_secs = 15
-  sqs.queue_url = "<%= @sqs_url %>"
+  sqs.queue_url = "<%= sqs_url %>"
 
-[transforms.parsing]
+[transforms.parse_<%= key %>]
   type = "remap"
-  inputs = ["s3_akamai_static"]
+  inputs = ["s3_akamai_<%= key %>"]
   source = '''
   . = parse_json!(string!(.message))
   .turnAroundTimeSec = to_int!(.turnAroundTimeMSec) / 1000.0
   '''
 
-[transforms.logs2metrics-requests]
+[transforms.logs2metrics_<%= key %>]
   type = "log_to_metric"
-  inputs = ["parsing"]
+  inputs = ["parse_<%= key %>"]
 
-  [[transforms.logs2metrics-requests.metrics]]
+  [[transforms.logs2metrics_<%= key %>.metrics]]
     field = "cacheStatus"
     name = "requests_cache_count"
-    namespace = "akamai_static"
+    namespace = "akamai_<%= key %>"
     type = "counter"
-      [transforms.logs2metrics-requests.metrics.tags]
+      [transforms.logs2metrics_<%= key %>.metrics.tags]
       status_code = "{{statusCode}}"
       cached = "{{cacheStatus}}"
       host = "{{reqHost}}"
 
-  [[transforms.logs2metrics-requests.metrics]]
+  [[transforms.logs2metrics_<%= key %>.metrics]]
     field = "bytes"
     name = "requests_bytes"
-    namespace = "akamai_static"
+    namespace = "akamai_<%= key %>"
     type = "counter"
     increment_by_value = true
-      [transforms.logs2metrics-requests.metrics.tags]
+      [transforms.logs2metrics_<%= key %>.metrics.tags]
       status_code = "{{statusCode}}"
       cached = "{{cacheStatus}}"
       host = "{{reqHost}}"
 
-  [[transforms.logs2metrics-requests.metrics]]
+  [[transforms.logs2metrics_<%= key %>.metrics]]
     field = "turnAroundTimeSec"
     name = "turnaround_time_sec"
-    namespace = "akamai_static"
+    namespace = "akamai_<%= key %>"
     type = "histogram"
-      [transforms.logs2metrics-requests.metrics.tags]
+      [transforms.logs2metrics_<%= key %>.metrics.tags]
       status_code = "{{statusCode}}"
       cached = "{{cacheStatus}}"
       host = "{{reqHost}}"
 
+<% end %>
+
 [sinks.prometheus_exporter]
   type = "prometheus_exporter"
   inputs = ["vector_metrics", "logs2metrics*"]