google-cloud-dataflow - Datadog を使用した Dataflow パイプラインのモニタリング

Question

Datadog を使用して GCP Dataflow パイプラインを監視し、ビルトイン メトリックと Beam カスタム メトリックを抽出するソリューションを探しています。現在、Datadog は他の GCP サービスの統合を提供していますが、Dataflow の統合は提供していません。誰かが同様の作業を行い、これをカスタム ソリューションとして構築する方法を共有できますか?

Answer 1

今のところ、2 つの可能性しか考えていません。

1. google-cloud-clients/google-cloud-monitoring のクライアントを使用して GCP カスタム メトリクスを使用し、Stackdriver と Datadog を統合する
2. クラウドにデプロイされた datadog エージェントを使用し、Datadog StatsD クライアント (Java、Python、Go) を使用して接続します。

---

1. GCP カスタム メトリクス https://cloud.google.com/monitoring/custom-metrics/creating-metrics と GCP との datadog 統合を 使用します https://www.datadoghq.com/product/integrations/#cat-google-cloud

   final MetricServiceClient client = MetricServiceClient.create();
   ProjectName name = ProjectName.of(projectId);
   
   MetricDescriptor descriptor = MetricDescriptor.newBuilder()
       .setType(metricType)
       .setDescription("This is a simple example of a custom metric.")
       .setMetricKind(MetricDescriptor.MetricKind.GAUGE)
       .setValueType(MetricDescriptor.ValueType.DOUBLE)
       .build();
   
   CreateMetricDescriptorRequest request = CreateMetricDescriptorRequest.newBuilder()
       .setName(name.toString())
       .setMetricDescriptor(descriptor)
       .build();
   
   client.createMetricDescriptor(request);
   

2. Datadog statsd クライアント、Java one - https://github.com/DataDog/java-dogstatsd-clientを使用して、Datadog エージェントを GCP にデプロイし、それを介して接続できるようにします。kubernetes でサンプルします。 https://docs.datadoghq.com/tracing/setup/kubernetes/#deploy-agent-daemonset

   import com.timgroup.statsd.ServiceCheck;
   import com.timgroup.statsd.StatsDClient;
   import com.timgroup.statsd.NonBlockingStatsDClient;
   
   public class Foo {
   
     private static final StatsDClient statsd = new NonBlockingStatsDClient(
       "my.prefix",                          /* prefix to any stats; may be null or empty string */
       "statsd-host",                        /* common case: localhost */
       8125,                                 /* port */
       new String[] {"tag:value"}            /* Datadog extension: Constant tags, always applied */
     );
   
     public static final void main(String[] args) {
       statsd.incrementCounter("foo");
       statsd.recordGaugeValue("bar", 100);
       statsd.recordGaugeValue("baz", 0.01); /* DataDog extension: support for floating-point gauges */
       statsd.recordHistogramValue("qux", 15);     /* DataDog extension: histograms */
       statsd.recordHistogramValue("qux", 15.5);   /* ...also floating-point */
       statsd.recordDistributionValue("qux", 15);     /* DataDog extension: global distributions */
       statsd.recordDistributionValue("qux", 15.5);   /* ...also floating-point */
   
       ServiceCheck sc = ServiceCheck
             .builder()
             .withName("my.check.name")
             .withStatus(ServiceCheck.Status.OK)
             .build();
       statsd.serviceCheck(sc); /* Datadog extension: send service check status */
   
       /* Compatibility note: Unlike upstream statsd, DataDog expects execution times to be a
        * floating-point value in seconds, not a millisecond value. This library
        * does the conversion from ms to fractional seconds.
        */
       statsd.recordExecutionTime("bag", 25, "cluster:foo"); /* DataDog extension: cluster tag */
     }
   }
   

   kubernetes の datadog deployment.yaml

   apiVersion: extensions/v1beta1
   kind: DaemonSet
   metadata:
     name: datadog-agent
   spec:
     template:
       metadata:
         labels:
           app: datadog-agent
         name: datadog-agent
       spec:
         serviceAccountName: datadog-agent
         containers:
         - image: datadog/agent:latest
           imagePullPolicy: Always
           name: datadog-agent
           ports:
             - containerPort: 8125
               # hostPort: 8125
               name: dogstatsdport
               protocol: UDP
             - containerPort: 8126
               # hostPort: 8126
               name: traceport
               protocol: TCP
           env:
             - name: DD_APM_ENABLED
               value: "true"
             - name: DD_API_KEY
               value: "<YOUR_API_KEY>"
             - name: DD_COLLECT_KUBERNETES_EVENTS
               value: "true"
             - name: DD_LEADER_ELECTION
               value: "true"
             - name: KUBERNETES
               value: "yes"
             - name: DD_KUBERNETES_KUBELET_HOST
               valueFrom:
                 fieldRef:
                   fieldPath: status.hostIP
           resources:
             requests:
               memory: "256Mi"
               cpu: "200m"
             limits:
               memory: "256Mi"
               cpu: "200m"
           volumeMounts:
             - name: dockersocket
               mountPath: /var/run/docker.sock
             - name: procdir
               mountPath: /host/proc
               readOnly: true
             - name: cgroups
               mountPath: /host/sys/fs/cgroup
               readOnly: true
           livenessProbe:
             exec:
               command:
               - ./probe.sh
             initialDelaySeconds: 15
             periodSeconds: 5
         volumes:
           - hostPath:
               path: /var/run/docker.sock
             name: dockersocket
           - hostPath:
               path: /proc
             name: procdir
           - hostPath:
               path: /sys/fs/cgroup
             name: cgroups
   

現在調査中のため、どうすればよいかわかりません。