enhancement(checks ipc): add new check data types and harden the protocol

lib/protos/datadog/build.rs

@@ -150,16 +150,21 @@ fn main() {
 
     tonic_prost_build::configure()
         .build_server(true)
+        .build_client(true)
         .include_file("checks.mod.rs")
         .compile_protos(
             &[
                 "proto/checks/v1/checks.proto",
+                "proto/checks/v1/check_runner.proto",
                 "proto/checks/v1/metric.proto",
                 "proto/checks/v1/log.proto",
                 "proto/checks/v1/service_check.proto",
                 "proto/checks/v1/event.proto",
+                "proto/checks/v1/sketch.proto",
+                "proto/checks/v1/event_platform_event.proto",
+                "proto/checks/v1/histogram_bucket.proto",
             ],
             &["proto"],
         )
-        .expect("Failed to build gRPC service definitions for Checks IPC.");
+        .expect("Failed to build checks proto definitions.");
 }

lib/protos/datadog/proto/checks/v1/check_runner.proto

@@ -0,0 +1,106 @@
+syntax = "proto3";
+
+package datadog.checks.v1;
+
+import "checks/v1/checks.proto";
+
+// Known protocol versions
+enum ProtocolVersion {
+  PROTOCOL_VERSION_UNSPECIFIED = 0;
+  PROTOCOL_VERSION_V1 = 1;
+}
+
+// Handshake: client -> server
+message HandshakeRequest {
+  ProtocolVersion protocol_version = 1;
+  string client_id = 2;
+  string client_version = 3;
+}
+
+// Handshake: server -> client
+message HandshakeResponse {
+  ProtocolVersion protocol_version = 1;
+  string server_id = 2;
+  string server_version = 3;
+  bool accepted = 4;
+  string reject_reason = 5;
+}
+
+// A single config update the server streams to the client
+message StreamConfigResponse {
+  EnrichmentData enrichment = 1;
+  repeated CheckInstance checks = 2;
+  map<string, string> agent_config = 3;
+}
+
+message EnrichmentData {
+  string hostname = 1;
+  map<string, string> host_tags = 2;
+  string cluster_name = 3;
+  string agent_version = 4;
+  map<string, bytes> config_values = 5;
+  uint64 process_start_time = 6;
+  K8sConnectionInfo k8s_connection_info = 7;
+}
+
+message K8sConnectionInfo {
+  string api_server_url = 1;
+  string bearer_token = 2;
+}
+
+message CheckInstance {
+  string check_name = 1;
+  string instance_name = 2;
+  string id = 3;
+  // The per-instance YAML/JSON block (a single entry from the
+  // integration's `instances:` list).
+  bytes config = 4;
+  // The integration-level YAML/JSON `init_config:` block, shared
+  // across every instance of this integration. Required for any check
+  // that reads defaults at Configure time (credentials, base URLs,
+  // shared tags, etc.).
+  bytes init_config = 5;
+}
+
+// Batched check data: client -> server
+message SendCheckDataRequest {
+  uint64 sequence_id = 1;
+  string check_name = 2;
+  string instance_name = 3;
+  string check_id = 4;
+  repeated CheckData data = 5;
+}
+
+// Acknowledgement of a check-data batch: server -> client
+message SendCheckDataResponse {
+  uint64 sequence_id = 1;
+  bool success = 2;
+  string error = 3;
+}
+
+// Result of a completed check run: client -> server
+message SendCheckResultRequest {
+  string check_name = 1;
+  string instance_name = 2;
+  string check_id = 3;
+  string error = 4;
+}
+
+// Empty request/response types for RPCs that don't need a body
+message StreamConfigRequest {}
+message SendCheckResultResponse {}
+
+// gRPC service a check runner uses to hand its output to a host process
+service CheckRunner {
+  // Handshake — the first call after connecting.
+  rpc Handshake(HandshakeRequest) returns (HandshakeResponse);
+
+  // Check data submission with acknowledgement.
+  rpc SendCheckData(SendCheckDataRequest) returns (SendCheckDataResponse);
+
+  // Check result reporting.
+  rpc SendCheckResult(SendCheckResultRequest) returns (SendCheckResultResponse);
+
+  // Server-streamed config updates.
+  rpc StreamConfig(StreamConfigRequest) returns (stream StreamConfigResponse);
+}

lib/protos/datadog/proto/checks/v1/checks.proto

@@ -3,25 +3,22 @@ syntax = "proto3";
 package datadog.checks.v1;
 
 import "checks/v1/event.proto";
+import "checks/v1/event_platform_event.proto";
+import "checks/v1/histogram_bucket.proto";
 import "checks/v1/log.proto";
 import "checks/v1/metric.proto";
 import "checks/v1/service_check.proto";
+import "checks/v1/sketch.proto";
 
 message CheckData {
     oneof data {
         datadog.checks.v1.metric.Metric metric = 1;
         datadog.checks.v1.log.Log log = 2;
         datadog.checks.v1.service_check.ServiceCheck service_check = 3;
         datadog.checks.v1.event.Event event = 4;
+        // Pre-aggregated DDSketch for a distribution-type metric
+        datadog.checks.v1.sketch.Sketch sketch = 5;
+        datadog.checks.v1.event_platform_event.EventPlatformEvent event_platform_event = 6;
+        datadog.checks.v1.histogram_bucket.HistogramBucket histogram_bucket = 7;
     }
 }
-
-message SendCheckPayloadRequest {
-    repeated CheckData data = 1;
-}
-
-message SendCheckPayloadResponse {}
-
-service Checks {
-    rpc SendCheckPayload(SendCheckPayloadRequest) returns (SendCheckPayloadResponse);
-}

lib/protos/datadog/proto/checks/v1/event_platform_event.proto

@@ -0,0 +1,8 @@
+syntax = "proto3";
+
+package datadog.checks.v1.event_platform_event;
+
+message EventPlatformEvent {
+  string event_data = 1;
+  string event_type = 2;
+}

lib/protos/datadog/proto/checks/v1/histogram_bucket.proto

@@ -0,0 +1,48 @@
+syntax = "proto3";
+
+package datadog.checks.v1.histogram_bucket;
+
+// HistogramBucket carries a single pre-aggregated histogram bucket
+// produced by an OpenMetrics / Prometheus-style integration
+message HistogramBucket {
+    // Metric name (e.g. "http.request.duration.count").
+    string name = 1;
+
+    // Number of observations counted by this bucket.
+    int64 value = 2;
+
+    // Bucket bounds. For point observations (single-value buckets),
+    // both bounds carry the same value.
+    double lower_bound = 3;
+    double upper_bound = 4;
+
+    // True when value increases monotonically across calls and the
+    // receiver should compute per-bucket deltas client-side. False
+    // when the producer already emits deltas.
+    bool monotonic = 5;
+
+    // Hostname this bucket was produced on. Empty means "use the
+    // receiving agent's default hostname".
+    string hostname = 6;
+
+    // Tags in `key:value` form.
+    repeated string tags = 7;
+
+    // When `monotonic=true`, indicates the receiver should emit the
+    // first observed value rather than wait for a second to compute a
+    // delta. Producers set this when they know the bucket started from
+    // zero recently (e.g. a freshly-scraped target).
+    bool flush_first_value = 8;
+
+    // true = several buckets share this metric context, told apart only by
+    // their bounds
+    //
+    // false = one bucket per context with bounds encoded in the tags (default)
+    //
+    // For monotonic buckets this selects whether the receiver keys delta 
+    // tracking by context alone or by context and bounds.
+    //
+    // Python / OpenMetrics checks always emit false. Go core checks that
+    // submit whole histograms set it true.
+    bool multiple_buckets_per_context = 9;
+}

lib/protos/datadog/proto/checks/v1/log.proto

@@ -12,7 +12,33 @@ enum LogLevel {
     LOG_LEVEL_CRITICAL = 50;
 }
 
+// Log carries a single log record produced by an integration check.
 message Log {
     string message = 1;
     LogLevel level = 2;
+
+    // Integration source name (e.g. "mysql", "http_check"). Used by the
+    // logs intake to route and tag.
+    string source = 3;
+
+    // Hostname the log was produced on. Empty means "use the receiving
+    // agent's default hostname".
+    string hostname = 4;
+
+    // Service the log relates to. Surfaces as `service` in the intake.
+    string service = 5;
+
+    // Log-level tags, in `key:value` form (matches the agent's wire
+    // format used elsewhere). Combined with host / integration tags
+    // downstream.
+    repeated string tags = 6;
+
+    // Unix-nanosecond timestamp at which the producer believes the log
+    // event happened. Zero means "stamp at receive time".
+    int64 timestamp = 7;
+
+    // Free-form additional structured fields. Each value is a JSON-encoded
+    // string so producers can convey arbitrary scalar / array / object
+    // shapes through a stable wire type.
+    map<string, string> additional_properties = 8;
 }

lib/protos/datadog/proto/checks/v1/metric.proto

@@ -8,6 +8,8 @@ enum MetricType {
     METRIC_TYPE_RATE = 2;
     METRIC_TYPE_GAUGE = 3;
     METRIC_TYPE_HISTOGRAM = 4;
+    METRIC_TYPE_MONOTONIC_COUNT = 5;
+    METRIC_TYPE_HISTORATE = 6;
 }
 
 message Metric {

lib/protos/datadog/proto/checks/v1/sketch.proto

@@ -0,0 +1,42 @@
+syntax = "proto3";
+
+package datadog.checks.v1.sketch;
+
+// Sketch carries a pre-aggregated DDSketch produced by a check. Its bin
+// encoding matches the Datadog Agent's sketch payload so a receiver can
+// convert it to its internal sketch type directly.
+message Sketch {
+    // Metric name (e.g. "http.request.duration").
+    string name = 1;
+
+    // Tags in `key:value` form.
+    repeated string tags = 2;
+
+    // Hostname this sketch was produced on. Empty means "use the
+    // receiving agent's default hostname".
+    string hostname = 3;
+
+    // Unix-second timestamp at which the producer flushed the sketch.
+    int64 timestamp = 4;
+
+    // Flush interval the producer used; informational only — the
+    // receiving aggregator stamps its own flush boundary.
+    uint64 interval_secs = 5;
+
+    // The DDSketch body itself, in the agent's bin-encoded shape.
+    Dogsketch dogsketch = 6;
+}
+
+// Dogsketch is the bin-encoded DDSketch body. Bin keys (k) and counts
+// (n) are parallel arrays: k[i] is the bin index and n[i] the number of
+// observations in that bin.
+message Dogsketch {
+    int64 ts = 1;
+    int64 cnt = 2;
+    double min = 3;
+    double max = 4;
+    double avg = 5;
+    double sum = 6;
+    repeated sint32 k = 7;
+    repeated uint32 n = 8;
+}