Add native OpenTelemetry support for deploy logs

Ship all deploy output and structured deploy events to an OTLP HTTP
endpoint. When configured via `otel.endpoint` in deploy.yml, Kamal
streams every line of stdout/stderr to the collector and emits
deploy.start/deploy.complete events with version, destination, runtime,
and performer metadata.

Resource attributes set automatically:
  - service.name: "kamal"
  - service.namespace: <service name from deploy.yml>
  - deployment.environment.name: <destination>
  - deploy.performer: <git user>

Usage in deploy.yml:

  otel:
    endpoint: http://otel-gateway:4318

Logs flow through the existing OTLP logs pipeline on the collector —
no new ports or receiver config needed.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: lewispb

lib/kamal/cli/main.rb

@@ -19,6 +19,8 @@ def setup
   option :skip_push, aliases: "-P", type: :boolean, default: false, desc: "Skip image build and push"
   option :no_cache, type: :boolean, default: false, desc: "Build without using Docker's build cache"
   def deploy(boot_accessories: false)
+    KAMAL.otel_event("deploy.start")
+
     runtime = print_runtime do
       invoke_options = deploy_options
 
@@ -48,13 +50,21 @@ def deploy(boot_accessories: false)
       end
     end
 
+    KAMAL.otel_event("deploy.complete", runtime: runtime.round.to_s)
     run_hook "post-deploy", secrets: true, runtime: runtime.round.to_s
+  rescue Exception => e
+    KAMAL.otel_event("deploy.failed", error: e.message)
+    raise
+  ensure
+    KAMAL.otel_shutdown
   end
 
   desc "redeploy", "Deploy app to servers without bootstrapping servers, starting kamal-proxy and pruning"
   option :skip_push, aliases: "-P", type: :boolean, default: false, desc: "Skip image build and push"
   option :no_cache, type: :boolean, default: false, desc: "Build without using Docker's build cache"
   def redeploy
+    KAMAL.otel_event("deploy.start", command: "redeploy")
+
     runtime = print_runtime do
       invoke_options = deploy_options
 
@@ -76,12 +86,20 @@ def redeploy
       end
     end
 
+    KAMAL.otel_event("deploy.complete", runtime: runtime.round.to_s, command: "redeploy")
     run_hook "post-deploy", secrets: true, runtime: runtime.round.to_s
+  rescue => e
+    KAMAL.otel_event("deploy.failed", error: e.message, command: "redeploy")
+    raise
+  ensure
+    KAMAL.otel_shutdown
   end
 
   desc "rollback [VERSION]", "Rollback app to VERSION"
   def rollback(version)
     rolled_back = false
+    KAMAL.otel_event("deploy.start", command: "rollback", rollback_version: version)
+
     runtime = print_runtime do
       with_lock do
         invoke_options = deploy_options
@@ -100,7 +118,13 @@ def rollback(version)
       end
     end
 
+    KAMAL.otel_event("deploy.complete", runtime: runtime.round.to_s, command: "rollback") if rolled_back
     run_hook "post-deploy", secrets: true, runtime: runtime.round.to_s if rolled_back
+  rescue => e
+    KAMAL.otel_event("deploy.failed", error: e.message, command: "rollback")
+    raise
+  ensure
+    KAMAL.otel_shutdown
   end
 
   desc "details", "Show details about all containers"

lib/kamal/commander.rb

@@ -4,7 +4,7 @@
 
 class Kamal::Commander
   attr_accessor :verbosity, :holding_lock, :connected
-  attr_reader :specific_roles, :specific_hosts
+  attr_reader :specific_roles, :specific_hosts, :otel_shipper
   delegate :hosts, :roles, :primary_host, :primary_role, :roles_on, :app_hosts, :proxy_hosts, :accessory_hosts, to: :specifics
 
   def initialize
@@ -142,6 +142,19 @@ def with_verbosity(level)
     SSHKit.config.output_verbosity = old_level
   end
 
+  def otel_enabled?
+    @otel_shipper.present?
+  end
+
+  def otel_event(name, **attrs)
+    @otel_shipper&.event(name, **attrs)
+  end
+
+  def otel_shutdown
+    @otel_shipper&.shutdown
+    @otel_shipper = nil
+  end
+
   def holding_lock?
     self.holding_lock
   end
@@ -161,6 +174,23 @@ def configure_sshkit_with(config)
       end
       SSHKit.config.command_map[:docker] = "docker" # No need to use /usr/bin/env, just clogs up the logs
       SSHKit.config.output_verbosity = verbosity
+
+      configure_otel_with(config)
+    end
+
+    def configure_otel_with(config)
+      return unless config.otel.enabled?
+
+      @otel_shipper = Kamal::OtelShipper.new(
+        endpoint: config.otel.endpoint,
+        service_namespace: config.otel.service_namespace,
+        environment: config.otel.environment,
+        version: config.version,
+        performer: `git config user.name`.chomp.presence || ENV["USER"]
+      )
+
+      $stdout = Kamal::TeeIo.new($stdout, @otel_shipper)
+      $stderr = Kamal::TeeIo.new($stderr, @otel_shipper)
     end
 
     def specifics

lib/kamal/configuration.rb

@@ -12,7 +12,7 @@ class Kamal::Configuration
   delegate :argumentize, :optionize, to: Kamal::Utils
 
   attr_reader :destination, :raw_config, :secrets
-  attr_reader :accessories, :aliases, :boot, :builder, :env, :logging, :proxy, :proxy_boot, :servers, :ssh, :sshkit, :registry
+  attr_reader :accessories, :aliases, :boot, :builder, :env, :logging, :otel, :proxy, :proxy_boot, :servers, :ssh, :sshkit, :registry
 
   include Validation
 
@@ -71,6 +71,7 @@ def initialize(raw_config, destination: nil, version: nil, validate: true)
     @env = Env.new(config: @raw_config.env || {}, secrets: secrets)
 
     @logging = Logging.new(logging_config: @raw_config.logging)
+    @otel = Otel.new(config: self)
     @proxy = Proxy.new(config: self, proxy_config: @raw_config.proxy, secrets: secrets)
     @proxy_boot = Proxy::Boot.new(config: self)
     @ssh = Ssh.new(config: self)

lib/kamal/configuration/docs/configuration.yml

@@ -206,6 +206,12 @@ boot:
 logging:
   ...
 
+# OpenTelemetry
+#
+# Ship deploy logs to an OTel endpoint, see kamal docs otel:
+otel:
+  ...
+
 # Aliases
 #
 # Alias configuration, see kamal docs alias:

lib/kamal/configuration/docs/otel.yml

@@ -0,0 +1,22 @@
+# OpenTelemetry
+#
+# Ship deploy logs to an OpenTelemetry-compatible endpoint via OTLP HTTP.
+#
+# When configured, Kamal streams all deploy output and emits structured
+# deploy events (start, complete, failed) to the specified endpoint.
+#
+# Logs are sent as OTLP log records with resource attributes:
+#   - service.name: "kamal"
+#   - service.namespace: <service name from config>
+#   - deployment.environment.name: <destination>
+
+# OTel options
+#
+# The options are specified under the otel key in the configuration file.
+otel:
+
+  # Endpoint
+  #
+  # The OTLP HTTP endpoint to send logs to (required to enable).
+  # This should be the base URL of the OTel collector, e.g. http://otel-gateway:4318
+  endpoint: http://otel-gateway:4318

lib/kamal/configuration/otel.rb

@@ -0,0 +1,32 @@
+class Kamal::Configuration::Otel
+  include Kamal::Configuration::Validation
+
+  attr_reader :otel_config
+
+  def initialize(config:)
+    @otel_config = config.raw_config.otel || {}
+    @service = config.service
+    @destination = config.destination
+    validate! otel_config unless otel_config.empty?
+  end
+
+  def enabled?
+    endpoint.present?
+  end
+
+  def endpoint
+    otel_config["endpoint"]
+  end
+
+  def service_namespace
+    @service
+  end
+
+  def environment
+    @destination
+  end
+
+  def to_h
+    otel_config
+  end
+end

lib/kamal/otel_shipper.rb

@@ -0,0 +1,120 @@
+require "net/http"
+require "json"
+require "uri"
+
+class Kamal::OtelShipper
+  BATCH_SIZE = 100
+  FLUSH_INTERVAL = 5 # seconds
+
+  def initialize(endpoint:, service_namespace:, environment:, version:, performer: nil)
+    @endpoint = URI("#{endpoint}/v1/logs")
+    @service_namespace = service_namespace
+    @environment = environment || "unknown"
+    @version = version
+    @performer = performer || ENV["USER"] || "unknown"
+    @buffer = Queue.new
+    @flush_mutex = Mutex.new
+    @running = true
+    @thread = start_flush_thread
+  end
+
+  def <<(str)
+    return self unless @running
+    str.to_s.each_line do |line|
+      stripped = line.chomp
+      @buffer << stripped unless stripped.empty?
+    end
+    flush if @buffer.size >= BATCH_SIZE
+    self
+  end
+
+  def event(name, **attributes)
+    attrs = attributes.map { |k, v| { key: k.to_s, value: { stringValue: v.to_s } } }
+    records = [ {
+      timeUnixNano: time_ns,
+      severityNumber: 9,
+      severityText: "INFO",
+      body: { stringValue: name },
+      attributes: attrs
+    } ]
+    ship_records(records)
+  end
+
+  def flush
+    @flush_mutex.synchronize do
+      lines = drain_buffer
+      return if lines.empty?
+
+      lines.each_slice(BATCH_SIZE) { |batch| ship_lines(batch) }
+    end
+  end
+
+  def shutdown
+    @running = false
+    @thread&.kill
+    flush
+  end
+
+  private
+    def start_flush_thread
+      Thread.new do
+        while @running
+          sleep FLUSH_INTERVAL
+          flush
+        end
+      end
+    end
+
+    def drain_buffer
+      lines = []
+      lines << @buffer.pop(true) until @buffer.empty?
+      lines
+    rescue ThreadError
+      lines
+    end
+
+    def ship_lines(lines)
+      records = lines.map do |line|
+        {
+          timeUnixNano: time_ns,
+          severityNumber: 9,
+          severityText: "INFO",
+          body: { stringValue: line }
+        }
+      end
+      ship_records(records)
+    end
+
+    def ship_records(records)
+      payload = {
+        resourceLogs: [ {
+          resource: { attributes: resource_attributes },
+          scopeLogs: [ { logRecords: records } ]
+        } ]
+      }
+
+      http = Net::HTTP.new(@endpoint.host, @endpoint.port)
+      http.use_ssl = @endpoint.scheme == "https"
+      http.open_timeout = 2
+      http.read_timeout = 5
+      req = Net::HTTP::Post.new(@endpoint.path, "Content-Type" => "application/json")
+      req.body = JSON.generate(payload)
+      http.request(req)
+    rescue
+      # Best effort — never fail the deploy
+    end
+
+    def resource_attributes
+      [
+        { key: "service.name", value: { stringValue: "kamal" } },
+        { key: "service.namespace", value: { stringValue: @service_namespace } },
+        { key: "service.version", value: { stringValue: @version } },
+        { key: "deployment.environment.name", value: { stringValue: @environment } },
+        { key: "deploy.performer", value: { stringValue: @performer } }
+      ]
+    end
+
+    def time_ns
+      (Time.now.to_f * 1_000_000_000).to_i.to_s
+    end
+end

lib/kamal/tee_io.rb

@@ -0,0 +1,41 @@
+class Kamal::TeeIo
+  def initialize(original, shipper)
+    @original = original
+    @shipper = shipper
+  end
+
+  def write(str)
+    @shipper << str
+    @original.write(str)
+  end
+
+  def <<(str)
+    @shipper << str
+    @original << str
+    self
+  end
+
+  def puts(*args)
+    str = args.empty? ? "\n" : args.map(&:to_s).join("\n") + "\n"
+    @shipper << str
+    @original.puts(*args)
+  end
+
+  def print(*args)
+    str = args.join
+    @shipper << str
+    @original.print(*args)
+  end
+
+  def flush
+    @original.flush if @original.respond_to?(:flush)
+  end
+
+  def method_missing(method, *args, &block)
+    @original.send(method, *args, &block)
+  end
+
+  def respond_to_missing?(method, include_private = false)
+    @original.respond_to?(method, include_private)
+  end
+end