Merge pull request #799 from saschagrunert/telemetry-link

Propagate existing telemetry spans to server

Author: openshift-merge-robot

.golangci.yml

@@ -95,7 +95,7 @@ linters:
     # - wsl
 linters-settings:
   funlen:
-    lines: 90
+    lines: 100
     statements: 50
   varnamelen:
     min-name-length: 1

Cargo.lock

@@ -5,6 +5,7 @@ interface Conmon {
     # Version
     struct VersionRequest {
         verbose @0 :Bool;
+        metadata @1 :Data;
     }
 
     struct VersionResponse {
@@ -17,6 +18,7 @@ interface Conmon {
         rustVersion @6 :Text;
         cargoVersion @7 :Text;
         cargoTree @8 :Text;
+        metadata @9 :Data;
     }
 
     version @0 (request: VersionRequest) -> (response: VersionResponse);
@@ -34,6 +36,7 @@ interface Conmon {
         cleanupCmd @7 :List(Text);
         globalArgs @8 :List(Text);
         commandArgs @9 :List(Text);
+        metadata @10 :Data;
     }
 
     struct LogDriver {
@@ -65,6 +68,7 @@ interface Conmon {
         timeoutSec @1 :UInt64;
         command @2 :List(Text);
         terminal @3 :Bool;
+        metadata @4 :Data;
     }
 
     struct ExecSyncContainerResponse {
@@ -83,6 +87,7 @@ interface Conmon {
         socketPath @1 :Text;
         execSessionId @2 :Text;
         stopAfterStdinEof @3 :Bool;
+        metadata @4 :Data;
     }
 
     struct AttachResponse {
@@ -94,6 +99,7 @@ interface Conmon {
     # ReopenLog
     struct ReopenLogRequest {
         id @0 :Text;
+        metadata @1 :Data;
     }
 
     struct ReopenLogResponse {
@@ -107,6 +113,7 @@ interface Conmon {
         id @0 :Text; # container identifier
         width @1 :UInt16; # columns in characters
         height @2 :UInt16; # rows in characters
+        metadata @3 :Data;
     }
 
     struct SetWindowSizeResponse {

conmon-rs/common/proto/conmon.capnp

@@ -29,6 +29,7 @@ prctl = "1.0.0"
 regex = "1.6.0"
 sendfd = { version = "0.4.3", features = ["tokio"] }
 serde = { version = "1.0.145", features = ["derive"] }
+serde_json = "1.0.86"
 shadow-rs = "=0.16.1"
 strum = { version = "0.24.1", features = ["derive"] }
 tempfile = "3.3.0"

conmon-rs/server/Cargo.toml

@@ -3,6 +3,7 @@ use crate::{
     container_io::{ContainerIO, SharedContainerIO},
     container_log::ContainerLog,
     server::Server,
+    telemetry::Telemetry,
     version::Version,
 };
 use anyhow::format_err;
@@ -68,8 +69,9 @@ impl conmon::Server for Server {
         debug!("Got a version request");
 
         let req = pry!(pry!(params.get()).get_request());
-        let version = Version::new(req.get_verbose());
+        pry_err!(Telemetry::set_parent_context(pry!(req.get_metadata())));
 
+        let version = Version::new(req.get_verbose());
         let mut response = results.get().init_response();
         response.set_process_id(process::id());
         response.set_version(version.version());
@@ -91,6 +93,8 @@ impl conmon::Server for Server {
         mut results: conmon::CreateContainerResults,
     ) -> Promise<(), capnp::Error> {
         let req = pry!(pry!(params.get()).get_request());
+        pry_err!(Telemetry::set_parent_context(pry!(req.get_metadata())));
+
         let id = pry!(req.get_id()).to_string();
         let cleanup_cmd: Vec<String> = pry!(pry!(req.get_cleanup_cmd())
             .iter()
@@ -180,6 +184,8 @@ impl conmon::Server for Server {
         mut results: conmon::ExecSyncContainerResults,
     ) -> Promise<(), capnp::Error> {
         let req = pry!(pry!(params.get()).get_request());
+        pry_err!(Telemetry::set_parent_context(pry!(req.get_metadata())));
+
         let id = pry!(req.get_id()).to_string();
         let timeout = req.get_timeout_sec();
 
@@ -262,6 +268,8 @@ impl conmon::Server for Server {
         _: conmon::AttachContainerResults,
     ) -> Promise<(), capnp::Error> {
         let req = pry!(pry!(params.get()).get_request());
+        pry_err!(Telemetry::set_parent_context(pry!(req.get_metadata())));
+
         let container_id = pry_err!(req.get_id());
 
         let span = new_root_span!("attach_container", container_id);
@@ -300,6 +308,8 @@ impl conmon::Server for Server {
         _: conmon::ReopenLogContainerResults,
     ) -> Promise<(), capnp::Error> {
         let req = pry!(pry!(params.get()).get_request());
+        pry_err!(Telemetry::set_parent_context(pry!(req.get_metadata())));
+
         let container_id = pry_err!(req.get_id());
 
         let span = new_root_span!("reopen_log_container", container_id);
@@ -322,6 +332,8 @@ impl conmon::Server for Server {
         _: conmon::SetWindowSizeContainerResults,
     ) -> Promise<(), capnp::Error> {
         let req = pry!(pry!(params.get()).get_request());
+        pry_err!(Telemetry::set_parent_context(pry!(req.get_metadata())));
+
         let container_id = pry_err!(req.get_id());
 
         let span = new_root_span!("set_window_size_container", container_id);

conmon-rs/server/src/rpc.rs

@@ -4,6 +4,7 @@ use clap::crate_name;
 use nix::unistd::gethostname;
 use opentelemetry::{
     global,
+    propagation::Extractor,
     runtime::Tokio,
     sdk::{
         propagation::TraceContextPropagator,
@@ -13,9 +14,9 @@ use opentelemetry::{
 };
 use opentelemetry_otlp::{ExportConfig, WithExportConfig};
 use opentelemetry_semantic_conventions::resource::{HOST_NAME, PROCESS_PID, SERVICE_NAME};
-use std::process;
-use tracing::Subscriber;
-use tracing_opentelemetry::OpenTelemetryLayer;
+use std::{collections::HashMap, process};
+use tracing::{Span, Subscriber};
+use tracing_opentelemetry::{OpenTelemetryLayer, OpenTelemetrySpanExt};
 use tracing_subscriber::registry::LookupSpan;
 
 /// The main structure of this module.
@@ -60,4 +61,32 @@ impl Telemetry {
     pub fn shutdown() {
         global::shutdown_tracer_provider();
     }
+
+    /// Set a new parent context from the provided slice data.
+    pub fn set_parent_context(slice: &'_ [u8]) -> Result<()> {
+        if slice.is_empty() {
+            // Make it a noop if no data is provided.
+            return Ok(());
+        }
+
+        let metadata = Metadata(serde_json::from_slice(slice).context("parse slice as JSON")?);
+        let ctx = global::get_text_map_propagator(|prop| prop.extract(&metadata));
+        Span::current().set_parent(ctx);
+
+        Ok(())
+    }
+}
+
+/// Additional telemetry metadata to carry.
+struct Metadata<'a>(HashMap<&'a str, &'a str>);
+
+impl<'a> Extractor for Metadata<'a> {
+    fn get(&self, key: &str) -> Option<&str> {
+        self.0.get(key).copied()
+    }
+
+    /// Collect all the keys from the MetadataMap.
+    fn keys(&self) -> Vec<&str> {
+        self.0.keys().copied().collect()
+    }
 }

conmon-rs/server/src/telemetry.rs

@@ -12,13 +12,17 @@ require (
 	github.com/opencontainers/runc v1.1.4
 	github.com/opencontainers/runtime-tools v0.9.1-0.20220714195903-17b3287fafb7
 	github.com/sirupsen/logrus v1.9.0
+	go.opentelemetry.io/otel v1.10.0
+	go.opentelemetry.io/otel/trace v1.10.0
 )
 
 require (
 	github.com/blang/semver v3.5.1+incompatible // indirect
 	github.com/coreos/go-systemd/v22 v22.4.0 // indirect
 	github.com/cyphar/filepath-securejoin v0.2.3 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect

go.mod

@@ -288,6 +288,11 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
 github.com/go-logr/logr v0.2.0/go.mod h1:z6/tIYblkpsD+a4lm/fGIIU9mZ+XfAiaFtq7xTgseGU=
+github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.2.3 h1:2DntVwHkVopvECVRSlL5PSo9eG+cAkDCuckLubN+rq0=
+github.com/go-logr/logr v1.2.3/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-openapi/jsonpointer v0.0.0-20160704185906-46af16f9f7b1/go.mod h1:+35s3my2LFTysnkMfxsJBAMHj/DoqoB9knIWoYG/Vk0=
 github.com/go-openapi/jsonpointer v0.19.2/go.mod h1:3akKfEdA7DF1sugOqz1dVQHBcuDBPKZGEoHC/NkiQRg=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
@@ -697,6 +702,10 @@ go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
+go.opentelemetry.io/otel v1.10.0 h1:Y7DTJMR6zs1xkS/upamJYk0SxxN4C9AqRd77jmZnyY4=
+go.opentelemetry.io/otel v1.10.0/go.mod h1:NbvWjCthWHKBEUMpf0/v8ZRZlni86PpGFEMA9pnQSnQ=
+go.opentelemetry.io/otel/trace v1.10.0 h1:npQMbR8o7mum8uF95yFbOEJffhs1sbCOfDh8zAJiH5E=
+go.opentelemetry.io/otel/trace v1.10.0/go.mod h1:Sij3YYczqAdz+EhmGhE6TpTxUO5/F/AzrK+kxfGqySM=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=