Add new route types

Author: estringana

datadog-sidecar/src/service/mod.rs

@@ -72,4 +72,5 @@ pub enum SidecarAction {
     AddTelemetryMetricPoint((String, f64, Vec<Tag>)),
     PhpComposerTelemetryFile(PathBuf),
     ClearQueueId,
+    AddRoute(Route),
 }

ddtelemetry/src/data/payloads.rs

@@ -95,3 +95,64 @@ pub enum LogLevel {
     Warn,
     Debug,
 }
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
+pub struct EndpointsPayload {
+    pub is_first: Option<bool>,
+    pub endpoints: Vec<Endpoint>,
+}
+
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Hash, Clone)]
+#[serde(rename_all = "UPPERCASE")]
+#[repr(C)]
+pub enum Method {
+    Get,
+    Post,
+    Put,
+    Delete,
+    Patch,
+    Head,
+    Options,
+    Trace,
+    Connect,
+    Other, //This is specified as "*" in the OpenAPI spec
+}
+
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Hash, Clone)]
+#[serde(rename_all = "UPPERCASE")]
+#[repr(C)]
+pub enum Authentication {
+    Jwt,
+    Basic,
+    Oauth,
+    Oidc,
+    ApiKey,
+    Session,
+    Mtls,
+    Saml,
+    Ldap,
+    Form,
+    Other,
+}
+
+#[derive(Serialize, Deserialize, Debug, Hash, PartialEq, Eq, Clone, Default)]
+pub struct Endpoint {
+    #[serde(default)]
+    pub r#type: Option<String>,
+    #[serde(default)]
+    pub method: Option<Method>,
+    #[serde(default)]
+    pub path: Option<String>,
+    pub operation_name: String,
+    pub resource_name: String,
+    #[serde(default)]
+    pub request_body_type: Option<Vec<String>>,
+    #[serde(default)]
+    pub response_body_type: Option<Vec<String>>,
+    #[serde(default)]
+    pub response_code: Option<Vec<i32>>,
+    #[serde(default)]
+    pub authentication: Option<Vec<Authentication>>,
+    #[serde(default)]
+    pub metadata: Option<serde_json::Value>,
+}

ddtelemetry/src/worker/mod.rs

@@ -7,7 +7,7 @@ pub mod store;
 
 use crate::{
     config::Config,
-    data::{self, Application, Dependency, Host, Integration, Log, Payload, Telemetry},
+    data::{self, Application, Dependency, Endpoint, Host, Integration, Log, Payload, Telemetry},
     metrics::{ContextKey, MetricBuckets, MetricContexts},
 };
 use ddcommon::Endpoint;
@@ -79,6 +79,7 @@ pub enum TelemetryActions {
     AddDependency(Dependency),
     AddIntegration(Integration),
     AddLog((LogIdentifier, Log)),
+    AddEndpoint(Endpoint),
     Lifecycle(LifecycleAction),
     #[serde(skip)]
     CollectStats(oneshot::Sender<TelemetryWorkerStats>),
@@ -110,6 +111,7 @@ struct TelemetryWorkerData {
     dependencies: store::Store<Dependency>,
     configurations: store::Store<data::Configuration>,
     integrations: store::Store<data::Integration>,
+    endpoints: store::Store<data::Endpoint>,
     logs: store::QueueHashMap<LogIdentifier, Log>,
     metric_contexts: MetricContexts,
     metric_buckets: MetricBuckets,
@@ -181,6 +183,8 @@ pub struct TelemetryWorkerStats {
     pub configurations_unflushed: u32,
     pub integrations_stored: u32,
     pub integrations_unflushed: u32,
+    pub endpoints_stored: u32,
+    pub endpoints_unflushed: u32,
     pub logs: u32,
     pub metric_contexts: u32,
     pub metric_buckets: MetricBucketStats,
@@ -197,6 +201,8 @@ impl Add for TelemetryWorkerStats {
             configurations_unflushed: self.configurations_unflushed + rhs.configurations_unflushed,
             integrations_stored: self.integrations_stored + rhs.integrations_stored,
             integrations_unflushed: self.integrations_unflushed + rhs.integrations_unflushed,
+            endpoints_stored: self.endpoints_stored + rhs.endpoints_stored,
+            endpoints_unflushed: self.endpoints_unflushed + rhs.endpoints_unflushed,
             logs: self.logs + rhs.logs,
             metric_contexts: self.metric_contexts + rhs.metric_contexts,
             metric_buckets: MetricBucketStats {
@@ -315,7 +321,7 @@ impl TelemetryWorker {
                     }
                 }
             }
-            AddConfig(_) | AddDependency(_) | AddIntegration(_) | Lifecycle(ExtendedHeartbeat) => {}
+            AddConfig(_) | AddDependency(_) | AddIntegration(_) | AddEndpoint(_) | Lifecycle(ExtendedHeartbeat) => {}
             Lifecycle(Stop) => {
                 if !self.data.started {
                     return BREAK;
@@ -372,6 +378,7 @@ impl TelemetryWorker {
             AddDependency(dep) => self.data.dependencies.insert(dep),
             AddIntegration(integration) => self.data.integrations.insert(integration),
             AddConfig(cfg) => self.data.configurations.insert(cfg),
+            AddEndpoint(endpoint) => self.data.endpoints.insert(endpoint),
             AddLog((identifier, log)) => {
                 let (l, new) = self.data.logs.get_mut_or_insert(identifier, log);
                 if !new {
@@ -424,6 +431,7 @@ impl TelemetryWorker {
                 self.data.dependencies.unflush_stored();
                 self.data.integrations.unflush_stored();
                 self.data.configurations.unflush_stored();
+                self.data.endpoints.unflush_stored();
 
                 let app_started = data::Payload::AppStarted(self.build_app_started());
                 match self.send_payload(&app_started).await {
@@ -516,6 +524,13 @@ impl TelemetryWorker {
                 },
             ))
         }
+        if self.data.endpoints.flush_not_empty() {
+            payloads.push(data::Payload::AppEndpointsChange(
+                data::AppEndpointsChange {
+                    endpoints: self.data.endpoints.unflushed().cloned().collect(),
+                },
+            ))
+        }
         payloads
     }
 
@@ -618,6 +633,9 @@ impl TelemetryWorker {
                 .data
                 .configurations
                 .removed_flushed(p.configuration.len()),
+            AppEndpointsChange(p) => {
+                self.data.endpoints.removed_flushed(p.endpoints.len())
+            }
             MessageBatch(batch) => {
                 for p in batch {
                     self.payload_sent_success(p);
@@ -722,6 +740,8 @@ impl TelemetryWorker {
             configurations_unflushed: self.data.configurations.len_unflushed() as u32,
             integrations_stored: self.data.integrations.len_stored() as u32,
             integrations_unflushed: self.data.integrations.len_unflushed() as u32,
+            endpoints_stored: self.data.endpoints.len_stored() as u32,
+            endpoints_unflushed: self.data.endpoints.len_unflushed() as u32,
             logs: self.data.logs.len() as u32,
             metric_contexts: self.data.metric_contexts.lock().len() as u32,
             metric_buckets: self.data.metric_buckets.stats(),
@@ -910,7 +930,7 @@ impl TelemetryWorkerHandle {
     }
 }
 
-/// How many dependencies/integrations/configs we keep in memory at most
+/// How many dependencies/integrations/configs/endpoints we keep in memory at most
 pub const MAX_ITEMS: usize = 5000;
 
 #[derive(Debug, Default, Clone, Copy)]
@@ -930,6 +950,7 @@ pub struct TelemetryWorkerBuilder {
     pub dependencies: store::Store<data::Dependency>,
     pub integrations: store::Store<data::Integration>,
     pub configurations: store::Store<data::Configuration>,
+    pub endpoints: store::Store<data::Endpoint>,
     pub native_deps: bool,
     pub rust_shared_lib_deps: bool,
     pub config: Config,
@@ -980,6 +1001,7 @@ impl TelemetryWorkerBuilder {
             dependencies: store::Store::new(MAX_ITEMS),
             integrations: store::Store::new(MAX_ITEMS),
             configurations: store::Store::new(MAX_ITEMS),
+            endpoints: store::Store::new(MAX_ITEMS),
             native_deps: true,
             rust_shared_lib_deps: false,
             config: Config::default(),
@@ -1010,6 +1032,7 @@ impl TelemetryWorkerBuilder {
                 dependencies: self.dependencies,
                 integrations: self.integrations,
                 configurations: self.configurations,
+                endpoints: self.endpoints,
                 logs: store::QueueHashMap::default(),
                 metric_contexts: contexts.clone(),
                 metric_buckets: MetricBuckets::default(),