Orphan telemetry for ecto spans originating from engine.ex in Oban

[olivermt]

Describe the bug Tons of UPDATE "public"."oban_producers" AS o0 SET "updated_at" = $1 WHERE (o0."uuid" = $2) as parentless spans Expected behavior Span from engine.ex should be respected and included as the parent span

Additional context Offending code: https://github.com/sorentwo/oban/blob/main/lib/oban/queue/engine.ex#L117

I suspect this is probably happening for all the functions in engine.ex as there seems to be nothing special about this function compared to all the other ones.

[bryannaegele]

This may be resolved in #49

[dvic]

@olivermt These queries will become "less" orphaned with #49 merged but it's still quite a lot of spans. Using #49 we actually ignored most of them in our setup using a sampler somewhere along these lines:

defmodule OpentelemetrySampler do
  @behaviour :otel_sampler

  defstruct [:opts]

  @impl true
  def setup(opts), do: %__MODULE__{opts: opts}

  @impl true
  def description(%__MODULE__{}), do: "OpentelemetrySampler"

  @impl true
  def should_sample(ctx, _trace_id, _links, span_name, _span_kind, attributes, _opts) do
    drop =
      String.ends_with?(span_name, ".repo.query:oban_producers") ||
        String.ends_with?(span_name, ".repo.query:oban_jobs") ||
        String.starts_with?(span_name, "Elixir.Oban.Plugins") ||
        String.starts_with?(span_name, "Elixir.Oban.Pro.Plugins") ||
        Map.get(attributes, :"db.statement") in ["begin", "commit"]

    decision =
      if drop do
        :drop
      else
        :record_and_sample
      end

    additional_attributes = []

    tracestate =
      OpenTelemetry.Tracer.current_span_ctx(ctx)
      |> OpenTelemetry.Span.tracestate()

    {decision, additional_attributes, tracestate}
  end
end

[olivermt]

My sampler already looks more or less like that actually :) We cut down some 8 milllion events per day in production with the above.

[tsloughter]

Is it possible to make certain types of Oban spans configurable whether the user wants them included or not? Instead of requiring them add a sampler?

I'm not a fan of samplers having to be used simply to say 'don't include all of these' -- esp since they apply to all spans created, so even ones completely unrelated to oban have to go through the same sampler, at this time.

[indrekj]

The problem is that these spans are created by Ecto automatically deep inside the oban code.

In ruby there's:

OpenTelemetry::Common::Utilities.untraced do
  # code
end

which basically creates a root span with sampling set to 0. So all the spans that are automatically created (activerecord, postgres, etc) inside the block get the sampled=false as well. The last time I checked then doing that was not possible in erlang opentelemetry. I don't think it is even in the standard but many languages have provided kind of similar behavior because of the same reasons as here.

[tsloughter]

Oh, interesting, we should probably add that and see if it should be in the spec.

[bryannaegele]

Nice. I can see how that could be helpful. We could maybe add a flag to the context in pdict that gets checked on span creation. 🤔

[aj-foster]

Hi folks 👋🏼

Is the custom sampler in the comment above still the best way to handle these traces?

[olivermt]

@aj-foster here is the current one we use (dominated by Oban still, but more or less the same approach yes):

Added bonus is a free howto on how to not log usernames and passwords in cleartext to honeycomb (🙈 ). Luckily it was caught in staging and HC has good api for deleting columns :P

defmodule Platform.OpentelemetrySampler do
  require OpenTelemetry.Tracer, as: Tracer
  require OpenTelemetry.Span, as: Span

  @moduledoc """
  `setup/1` takes a map of config you can utilize in your should_sample function

  `description/1` is used to produce a human readable description of the sampler in use.

  `should_sample/7` is where you do the actual sampling.
  The params are the following
  - OpenTelemetry.Ctx, the context of the whole trace
  - trace_id, the id of the current trace
  - links, list of span links (TODO ??)
  - span_name, the name of the current span
  - span_kind, the ??
  - attrs, the attributes for the current span
  - config, the return of the setup function

  `should_sample/7` needs to return a three-tuple:

  {
    drop | record_only | record_and_sample, # the sample decision
    [{"new_attrs", "here"}, ... {...}], #any new attrs
    trace_state #the current tracestate, see `get_tracestate`
  }
  """
  def setup(_) do
    %{}
  end

  def description(_), do: "PlatformSampler"

  @filter_list_db_statement [
    "begin",
    "commit",
    "SELECT pg_notify($1, payload) FROM json_array_elements_text($2::json) AS payload",
    "SELECT pg_try_advisory_xact_lock($1)",
    "select now()",
    ~s/UPDATE "public"."oban_jobs" AS o0 SET "state" = $1 FROM (SELECT so0."id" AS "id", so0."state" AS "state", so0."queue" AS "queue", so0."worker" AS "worker", so0."args" AS "args", so0."meta" AS "meta", so0."tags" AS "tags", so0."errors" AS "errors", so0."attempt" AS "attempt", so0."attempted_by" AS "attempted_by", so0."max_attempts" AS "max_attempts", so0."priority" AS "priority", so0."attempted_at" AS "attempted_at", so0."cancelled_at" AS "cancelled_at", so0."completed_at" AS "completed_at", so0."discarded_at" AS "discarded_at", so0."inserted_at" AS "inserted_at", so0."scheduled_at" AS "scheduled_at" FROM "public"."oban_jobs" AS so0 WHERE (so0."state" IN ('scheduled','retryable')) AND (NOT (so0."queue" IS NULL)) AND (so0."scheduled_at" <= $2) ORDER BY so0."id" LIMIT $3 FOR UPDATE SKIP LOCKED) AS s1 WHERE (o0."id" = s1."id")/,
    ~s/UPDATE "public"."oban_jobs" AS o0 SET "state" = $1, "attempted_at" = $2, "attempted_by" = $3, "attempt" = o0."attempt" + $4 WHERE (o0."id" IN (SELECT so0."id" FROM "public"."oban_jobs" AS so0 WHERE ((so0."state" = 'available') AND (so0."queue" = $5)) ORDER BY so0."priority", so0."scheduled_at", so0."id" LIMIT $6 FOR UPDATE SKIP LOCKED)) RETURNING o0."id", o0."state", o0."queue", o0."worker", o0."args", o0."meta", o0."tags", o0."errors", o0."attempt", o0."attempted_by", o0."max_attempts", o0."priority", o0."attempted_at", o0."cancelled_at", o0."completed_at", o0."discarded_at", o0."inserted_at", o0."scheduled_at"/,
    ~s/UPDATE "public"."oban_jobs" AS o0 SET "state" = $1, "discarded_at" = $2 FROM (SELECT so0."id" AS "id" FROM "public"."oban_jobs" AS so0 LEFT OUTER JOIN "public"."oban_producers" AS so1 ON array_length(so0."attempted_by", 1) = 2 AND (so1."uuid" = uuid (so0."attempted_by"[2])) WHERE (NOT (so0."queue" IS NULL) AND (so0."state" = 'executing')) AND (so1."uuid" IS NULL)) AS s1 WHERE (o0."id" = s1."id") AND (o0."attempt" >= o0."max_attempts")/,
    ~s/UPDATE "public"."oban_jobs" AS o0 SET "state" = $1 FROM (SELECT so0."id" AS "id" FROM "public"."oban_jobs" AS so0 LEFT OUTER JOIN "public"."oban_producers" AS so1 ON array_length(so0."attempted_by", 1) = 2 AND (so1."uuid" = uuid (so0."attempted_by"[2])) WHERE (NOT (so0."queue" IS NULL) AND (so0."state" = 'executing')) AND (so1."uuid" IS NULL)) AS s1 WHERE (o0."id" = s1."id") AND (o0."attempt" < o0."max_attempts")/,
    ~s/DELETE FROM "public"."oban_jobs" AS o0 USING (SELECT so0."id" AS "id" FROM "public"."oban_jobs" AS so0 WHERE ((so0."state" = 'completed') AND (so0."attempted_at" < $1)) OR ((so0."state" = 'cancelled') AND (so0."cancelled_at" < $2)) OR ((so0."state" = 'discarded') AND (so0."discarded_at" < $3)) LIMIT $4 FOR UPDATE SKIP LOCKED) AS s1 WHERE (o0."id" = s1."id")/,
    ~s/UPDATE "public"."oban_producers" AS o0 SET "updated_at" = $1 WHERE (o0."uuid" = $2)/,
    ~s/DELETE FROM "public"."oban_producers" AS o0 WHERE (((o0."uuid" != $1) AND (o0."updated_at" <= $2))) OR ((((o0."uuid" != $3) AND (o0."name" = $4)) AND (o0."queue" = $5)) AND (o0."updated_at" <= $6))/,
    ~s/SELECT DISTINCT o0."queue" FROM "public"."oban_jobs" AS o0 WHERE (o0."state" = 'available') AND (NOT (o0."queue" IS NULL))/,
    ~s/INSERT INTO "public"."oban_producers" ("meta","name","node","queue","running_ids","started_at","updated_at","uuid") VALUES ($1,$2,$3,$4,$5,$6,$7,$8)/,
    ~s/SELECT (o0."meta"#>'{"global_limit","tracked"}') FROM "public"."oban_producers" AS o0 WHERE (o0."queue" = $1) AND (NOT ((o0."meta"#>'{"global_limit","allowed"}') IS NULL))/,
    ~s/DELETE FROM "public"."oban_peers" AS o0 WHERE (o0."name" = $1) AND (o0."expires_at" < $2)/,
    ~s/INSERT INTO "public"."oban_peers" AS o0 ("expires_at","name","node","started_at") VALUES ($1,$2,$3,$4) ON CONFLICT ("name") DO UPDATE SET "expires_at" = $5/,
    ~s/INSERT INTO "public"."oban_peers" ("expires_at","name","node","started_at") VALUES ($1,$2,$3,$4) ON CONFLICT DO NOTHING/,
    ~s/SELECT key FROM UNNEST($1::int[]) key WHERE NOT pg_try_advisory_xact_lock($2, key)\n/,
    ~s/SELECT o0."uuid", o0."name", o0."node", o0."queue", o0."running_ids", o0."started_at", o0."updated_at", o0."meta" FROM "public"."oban_producers" AS o0 WHERE (o0."uuid" = $1) FOR UPDATE/,
    ~s/select 1/
  ]

  @filter_list_span_name [
    "Elixir.Oban.Plugins.Gossip process",
    "Elixir.Oban.Plugins.Stager process",
    "Elixir.Oban.Plugins.Pruner process"
  ]

  @filter_graphql_variables [
    "password",
    "token"
  ]
  def should_sample(
        ctx,
        _trace_id,
        _links,
        span_name,
        _span_kind,
        attrs,
        _config
      ) do
    {:no_drop, ctx, span_name, attrs}
    |> db_attrs_statement_should_drop()
    |> span_name_should_drop()
    |> gql_variable_atters_modify()
    |> case do
      {:drop, ctx, _span_name, _attrs} ->
        {:drop, [], tracestate(ctx)}

      {:no_drop, ctx, _span_name, new_attrs} ->
        {:record_and_sample, new_attrs, tracestate(ctx)}
    end
  end

  defp tracestate(ctx) do
    Tracer.current_span_ctx(ctx)
    |> Span.tracestate()
  end

  defp db_attrs_statement_should_drop(
         {_decision, ctx, span_name, %{"db.statement": statement} = attrs}
       )
       when statement in @filter_list_db_statement,
       do: {:drop, ctx, span_name, attrs}

  defp db_attrs_statement_should_drop(sample), do: sample

  defp gql_variable_atters_modify(
         {:no_drop, ctx, span_name, %{"graphql.request.variables" => variables} = attrs}
       ) do
    if String.contains?(variables, @filter_graphql_variables) do
      variables =
        variables
        |> Jason.decode()
        |> case do
          {:ok, variables} ->
            variables
            |> filter_values(@filter_graphql_variables)
            |> Jason.encode!()

          {:error, _reason} ->
            "[FILTERED]"
        end

      attrs = Map.put(attrs, "graphql.request.variables", variables)
      {:no_drop, ctx, span_name, attrs}
    else
      {:no_drop, ctx, span_name, attrs}
    end
  end

  defp gql_variable_atters_modify(sample), do: sample

  defp span_name_should_drop({_decision, ctx, span_name, attrs})
       when span_name in @filter_list_span_name do
    {:drop, ctx, span_name, attrs}
  end

  defp span_name_should_drop(sample), do: sample

  def filter_values(%{} = map, filter_params) do
    Enum.into(map, %{}, fn {k, v} ->
      if is_binary(k) and String.contains?(k, filter_params) do
        {k, "[FILTERED]"}
      else
        {k, filter_values(v, filter_params)}
      end
    end)
  end

  def filter_values([_ | _] = list, filter_params) do
    Enum.map(list, &filter_values(&1, filter_params))
  end

  def filter_values(other, _filter_params), do: other
end