SemanticSpec.hs

{-# LANGUAGE OverloadedStrings #-}

-- | Asserts that spans emitted by 'processOne' carry the wire names
-- defined by the OpenTelemetry messaging semantic conventions and the
-- shibuya-namespaced fallbacks. This is the guard that catches drift
-- between Shibuya's emitted attributes and the upstream spec — a future
-- rename in @hs-opentelemetry-semantic-conventions@ will break both
-- compilation (because @Shibuya.Telemetry.Semantic@ derives the strings
-- from typed @AttributeKey@s) and these assertions (because the wire
-- string changed).
module Shibuya.Telemetry.SemanticSpec (spec) where

import Data.Foldable (toList)
import Data.HashMap.Strict (HashMap)
import Data.HashMap.Strict qualified as HashMap
import Data.IORef (readIORef)
import Data.Int (Int64)
import Data.Text (Text)
import Effectful (runEff)
import OpenTelemetry.Attributes
  ( Attribute (..),
    PrimitiveAttribute (..),
    emptyAttributes,
    getAttributeMap,
    toAttribute,
  )
import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)
import OpenTelemetry.Trace.Core
  ( Event (..),
    ImmutableSpan (..),
    InstrumentationLibrary (..),
    createTracerProvider,
    emptyTracerProviderOptions,
    hotAttributes,
    hotEvents,
    hotName,
    makeTracer,
    shutdownTracerProvider,
    tracerOptions,
  )
import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)
import Shibuya.Adapter.Mock (listAdapter)
import Shibuya.Core.Ack (AckDecision (..))
import Shibuya.Core.AckHandle (AckHandle (..))
import Shibuya.Core.Ingested (Ingested (..))
import Shibuya.Core.Types (Envelope (..), MessageId (..))
import Shibuya.Runner.Metrics (ProcessorId (..))
import Shibuya.Runner.Supervised (runWithMetrics)
import Shibuya.Telemetry.Effect (runTracing)
import Test.Hspec

spec :: Spec
spec = describe "Shibuya.Telemetry.Semantic (wire-format)" $ do
  it "emits a process span with conventions-aligned attributes and events" $ do
    (processor, spansRef) <- inMemoryListExporter
    provider <- createTracerProvider [processor] emptyTracerProviderOptions
    let tracer = mkTestTracer provider

    runEff $ runTracing tracer $ do
      let envelope =
            Envelope
              { messageId = MessageId "m-1",
                cursor = Nothing,
                partition = Nothing,
                enqueuedAt = Nothing,
                traceContext = Nothing,
                attempt = Nothing,
                attributes = HashMap.empty,
                payload = ("hello" :: Text)
              }
          ingested =
            Ingested
              { envelope = envelope,
                ack = AckHandle (\_ -> pure ()),
                lease = Nothing
              }
          adapter = listAdapter [ingested]
          handler _ = pure AckOk
          procId = ProcessorId "test-proc"
      _ <- runWithMetrics 4 procId adapter handler
      pure ()

    _ <- shutdownTracerProvider provider (Just 5_000_000)
    spans <- readIORef spansRef
    case spans of
      [s] -> do
        hot <- readIORef (spanHot s)
        hotName hot `shouldBe` "test-proc process"
        let attrs = getAttributeMap (hotAttributes hot)
        attrs `shouldHaveTextAttribute` ("messaging.system", "shibuya")
        attrs `shouldHaveTextAttribute` ("messaging.message.id", "m-1")
        attrs `shouldHaveTextAttribute` ("messaging.destination.name", "test-proc")
        attrs `shouldHaveTextAttribute` ("messaging.operation.type", "process")
        attrs `shouldHaveTextAttribute` ("shibuya.ack.decision", "ack_ok")
        attrs `shouldHaveIntAttribute` ("shibuya.inflight.count", 1)
        attrs `shouldHaveIntAttribute` ("shibuya.inflight.max", 1)
        let evNames = map eventName (toList (appendOnlyBoundedCollectionValues (hotEvents hot)))
        evNames `shouldContain` ["shibuya.handler.started"]
        evNames `shouldContain` ["shibuya.handler.completed"]
      _ ->
        expectationFailure $ "expected exactly one span, got " <> show (length spans)

  it "applies envelope.attributes onto the framework span (P0 fix, plan 9 F1/F2)" $ do
    -- The adapter contributes broker-specific typed attributes via
    -- 'Envelope.attributes'. The framework's processOne span must carry
    -- them, and adapter-supplied keys must override framework defaults
    -- of the same name (here: messaging.system flips from "shibuya" to
    -- "kafka" because the adapter set it).
    (processor, spansRef) <- inMemoryListExporter
    provider <- createTracerProvider [processor] emptyTracerProviderOptions
    let tracer = mkTestTracer provider

    runEff $ runTracing tracer $ do
      let envelope =
            Envelope
              { messageId = MessageId "orders-2-42",
                cursor = Nothing,
                partition = Nothing,
                enqueuedAt = Nothing,
                traceContext = Nothing,
                attempt = Nothing,
                attributes =
                  HashMap.fromList
                    [ ("messaging.system", toAttribute ("kafka" :: Text)),
                      ( "messaging.kafka.destination.partition",
                        toAttribute (2 :: Int64)
                      ),
                      ( "messaging.kafka.message.offset",
                        toAttribute (42 :: Int64)
                      )
                    ],
                payload = ("hello" :: Text)
              }
          ingested =
            Ingested
              { envelope = envelope,
                ack = AckHandle (\_ -> pure ()),
                lease = Nothing
              }
          adapter = listAdapter [ingested]
          handler _ = pure AckOk
          procId = ProcessorId "orders-consumer"
      _ <- runWithMetrics 4 procId adapter handler
      pure ()

    _ <- shutdownTracerProvider provider (Just 5_000_000)
    spans <- readIORef spansRef
    case spans of
      [s] -> do
        hot <- readIORef (spanHot s)
        let attrs = getAttributeMap (hotAttributes hot)
        -- Adapter override wins.
        attrs `shouldHaveTextAttribute` ("messaging.system", "kafka")
        -- Adapter-typed attributes appear on the framework span.
        attrs `shouldHaveIntAttribute` ("messaging.kafka.destination.partition", 2)
        attrs `shouldHaveIntAttribute` ("messaging.kafka.message.offset", 42)
        -- Framework defaults still set where the adapter did not override.
        attrs `shouldHaveTextAttribute` ("messaging.destination.name", "orders-consumer")
        attrs `shouldHaveTextAttribute` ("messaging.operation.type", "process")
        attrs `shouldHaveTextAttribute` ("messaging.message.id", "orders-2-42")
      _ ->
        expectationFailure $
          "expected exactly one span, got " <> show (length spans)
  where
    mkTestTracer p =
      makeTracer
        p
        ( InstrumentationLibrary
            { libraryName = "shibuya-test",
              libraryVersion = "",
              librarySchemaUrl = "",
              libraryAttributes = emptyAttributes
            }
        )
        tracerOptions
    shouldHaveTextAttribute :: HashMap Text Attribute -> (Text, Text) -> Expectation
    shouldHaveTextAttribute attrs (k, expected) =
      case HashMap.lookup k attrs of
        Just (AttributeValue (TextAttribute v)) -> v `shouldBe` expected
        Just other ->
          expectationFailure $
            "attribute " <> show k <> " was not a Text: " <> show other
        Nothing ->
          expectationFailure $
            "attribute " <> show k <> " missing; have keys " <> show (HashMap.keys attrs)
    shouldHaveIntAttribute :: HashMap Text Attribute -> (Text, Int) -> Expectation
    shouldHaveIntAttribute attrs (k, expected) =
      case HashMap.lookup k attrs of
        Just (AttributeValue (IntAttribute v)) -> v `shouldBe` fromIntegral expected
        Just other ->
          expectationFailure $
            "attribute " <> show k <> " was not an Int: " <> show other
        Nothing ->
          expectationFailure $
            "attribute " <> show k <> " missing; have keys " <> show (HashMap.keys attrs)