Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update OpenTelemetryEcto attributes to conform with semantic conventions 1.27 #430

Draft
wants to merge 4 commits into
base: main
Choose a base branch
from
Draft

Update OpenTelemetryEcto attributes to conform with semantic conventions 1.27 #430

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
39c3318
Update otel versions
danschultzer Dec 13, 2024
6feb8ba
Use semantic conventions
danschultzer Dec 13, 2024
b8062a0
Move total time into measurements
danschultzer Dec 13, 2024
4779873
Use 1.27 semantic convention
danschultzer Dec 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
212 changes: 124 additions & 88 deletions instrumentation/opentelemetry_ecto/lib/opentelemetry_ecto.ex
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,15 +20,16 @@ defmodule OpentelemetryEcto do

require OpenTelemetry.Tracer

alias OpenTelemetry.SemConv.Incubating.{DBAttributes, Metrics.DBMetrics}
alias OpenTelemetry.SemConv.{ErrorAttributes, ServerAttributes}

@typedoc """
Option that you can pass to `setup/2`.
"""
@typedoc since: "1.3.0"
@type setup_option() ::
{:time_unit, System.time_unit()}
| {:span_prefix, String.t()}
| {:additional_attributes, %{String.t() => term()}}
| {:db_statement, :enabled | :disabled | (String.t() -> String.t())}
{:additional_attributes, %{String.t() => term()}}
| {:db_query, :enabled | :disabled | (String.t() -> String.t())}

@doc """
Attaches the `OpentelemetryEcto` handler to your repo events.
Expand All @@ -54,18 +55,12 @@ defmodule OpentelemetryEcto do

You may also supply the following options in the second argument:

* `:time_unit` - a time unit used to convert the values of query phase
timings, defaults to `:microsecond`. See `System.convert_time_unit/3`.
* `:span_prefix` - the first part of the span name.
Defaults to the concatenation of the event name with periods, such as
`"blog.repo.query"`. This will always be followed with a colon and the
source (the table name for SQL adapters). For example: `"blog.repo.query:users"`.
* `:additional_attributes` - additional attributes to include in the span. If there
are conflicts with default provided attributes, the ones provided with
this config will have precedence.
* `:db_statement` - `:disabled` (default), `:enabled`, or a function.
* `:db_query` - `:disabled` (default), `:enabled`, or a function.
Whether or not to include DB statements in the **span attributes** (as the
`db.statement` attribute).
`#{DBAttributes.db_query_text()}` attribute).
Optionally provide a function that takes a query string and returns a
sanitized version of it. This is useful for removing sensitive information from the
query string. Unless this option is `:enabled` or a function,
Expand All @@ -80,9 +75,9 @@ defmodule OpentelemetryEcto do

@doc false
def handle_event(
event,
_event,
measurements,
%{query: query, source: source, result: query_result, repo: repo, type: type},
%{query: query, source: source, result: query_result, repo: repo},
config
) do
# Doing all this even if the span isn't sampled so the sampler
Expand All @@ -91,56 +86,31 @@ defmodule OpentelemetryEcto do
total_time = measurements.total_time
end_time = :opentelemetry.timestamp()
start_time = end_time - total_time
database = repo.config()[:database]

url =
case repo.config()[:url] do
nil ->
# TODO: add port
URI.to_string(%URI{scheme: "ecto", host: repo.config()[:hostname]})
measurements = Map.put(measurements, :total_time, total_time)
repo_config = Keyword.take(repo.config(), [:database, :hostname, :port])

url ->
url
end

span_prefix =
case Keyword.fetch(config, :span_prefix) do
{:ok, prefix} -> prefix
:error -> Enum.join(event, ".")
end

span_suffix = if source != nil, do: ":#{source}", else: ""
span_name = span_prefix <> span_suffix

time_unit = Keyword.get(config, :time_unit, :microsecond)
additional_attributes = Keyword.get(config, :additional_attributes, %{})

db_type =
case type do
:ecto_sql_query -> :sql
_ -> type
end

# TODO: need connection information to complete the required attributes
# net.peer.name or net.peer.ip and net.peer.port
base_attributes = %{
"db.type": db_type,
source: source,
"db.instance": database,
"db.name": database,
"db.url": url,
"total_time_#{time_unit}s": System.convert_time_unit(total_time, :native, time_unit)
}

db_statement_config = Keyword.get(config, :db_statement, :disabled)
db_statement_config = Keyword.get(config, :db_query, :disabled)

attributes =
base_attributes
|> add_measurements(measurements, time_unit)
|> maybe_add_db_statement(db_statement_config, query)
|> maybe_add_db_system(repo.__adapter__())
# TODO: need connection information to complete the required attributes
# net.peer.name or net.peer.ip and net.peer.port
%{
unquote(DBAttributes.db_system()) => db_system(repo.__adapter__()),
unquote(DBAttributes.db_namespace()) => repo_config[:database],
unquote(ServerAttributes.server_address()) => repo_config[:hostname]
}
|> maybe_add_db_collection_name(source)
|> maybe_add_server_port(repo_config)
|> maybe_add_db_operation_name(repo.__adapter__(), query)
|> maybe_add_error_type(repo.__adapter__(), query_result)
|> maybe_add_db_query_text(db_statement_config, query)
|> add_measurements(measurements)
|> add_additional_attributes(additional_attributes)

span_name = span_name(attributes)

parent_context =
case OpentelemetryProcessPropagator.fetch_ctx(self()) do
:undefined ->
Expand Down Expand Up @@ -185,59 +155,125 @@ defmodule OpentelemetryEcto do

defp format_error(_), do: ""

defp add_measurements(attributes, measurements, time_unit) do
measurements
|> Enum.reduce(attributes, fn
{k, v}, acc
when not is_nil(v) and k in [:decode_time, :query_time, :queue_time, :idle_time] ->
Map.put(
acc,
String.to_atom("#{k}_#{time_unit}s"),
System.convert_time_unit(v, :native, time_unit)
)

_, acc ->
acc
end)
@db_systems [
{Ecto.Adapters.Postgres, DBAttributes.db_system_values().postgresql},
{Ecto.Adapters.MyXQL, DBAttributes.db_system_values().mysql},
{Ecto.Adapters.SQLite3, DBAttributes.db_system_values().sqlite},
{Ecto.Adapters.Tds, DBAttributes.db_system_values().mssql}
]

for {adapter, system} <- @db_systems do
defp db_system(unquote(adapter)), do: unquote(system)
end

defp maybe_add_db_statement(attributes, :enabled, query) do
Map.put(attributes, :"db.statement", query)
# NOTE: This is the catch-all clause where we use other_sql as the db.system value, but it may not be a SQL based database.
defp db_system(_), do: unquote(DBAttributes.db_system_values().other_sql)

defp maybe_add_db_collection_name(attributes, nil), do: attributes

defp maybe_add_db_collection_name(attributes, source) do
Map.put(attributes, unquote(DBAttributes.db_collection_name()), source)
end

defp maybe_add_db_statement(attributes, :disabled, _query) do
attributes
defp maybe_add_server_port(attributes, repo_config) do
case Keyword.has_key?(repo_config, :port) do
false -> attributes
true -> Map.put(attributes, unquote(ServerAttributes.server_port()), repo_config[:port])
end
end

defp maybe_add_db_statement(attributes, sanitizer, query) when is_function(sanitizer, 1) do
Map.put(attributes, :"db.statement", sanitizer.(query))
defp maybe_add_db_operation_name(attributes, adapter, query) do
case get_db_operation_name(adapter, query) do
nil -> attributes
operation_name -> Map.put(attributes, unquote(DBAttributes.db_operation_name()), operation_name)
end
end

defp maybe_add_db_statement(attributes, _, _query) do
attributes
# NOTE: Postgres does set a `:command` attribute on the result, but since there is no command for the
# error struct we will parse it all the same here.
defp get_db_operation_name(Ecto.Adapters.Postgres, query), do: sql_command(query)
defp get_db_operation_name(Ecto.Adapters.MyXQL, query), do: sql_command(query)
defp get_db_operation_name(Ecto.Adapters.SQLite3, query), do: sql_command(query)
defp get_db_operation_name(Ecto.Adapters.Tds, query), do: sql_command(query)
defp get_db_operation_name(_, _), do: nil

defp sql_command(query) when is_binary(query) do
query
|> String.split(" ", trim: true)
|> sql_command()
end

defp maybe_add_db_system(attributes, Ecto.Adapters.Postgres) do
Map.put(attributes, :"db.system", :postgresql)
@sql_commands ~w(select insert update delete begin commit)

defp sql_command([raw_command | _rest]) do
case String.downcase(raw_command) do
command when command in @sql_commands -> raw_command
_ -> nil
end
end

defp maybe_add_db_system(attributes, Ecto.Adapters.MyXQL) do
Map.put(attributes, :"db.system", :mysql)
defp maybe_add_error_type(attributes, _adapter, {:ok, _}), do: attributes

defp maybe_add_error_type(attributes, adapter, {:error, error}) do
case get_error_type(adapter, error) do
nil -> attributes
error_type -> Map.put(attributes, unquote(ErrorAttributes.error_type()), error_type)
end
end

defp maybe_add_db_system(attributes, Ecto.Adapters.SQLite3) do
Map.put(attributes, :"db.system", :sqlite)
defp get_error_type(Ecto.Adapters.Postgres, %{postgres: %{code: code}}), do: code
defp get_error_type(Ecto.Adapters.MyXQL, %{postgres: %{name: name}}), do: name
# NOTE: Exqlite.Error does not have an error type
# TODO: Normalize error type from the error message?
defp get_error_type(Ecto.Adapters.SQLite3, _), do: nil
defp get_error_type(Ecto.Adapters.Tds, %{mssql: %{number: number}}), do: number
defp get_error_type(_adapter, _), do: nil

@measurements [
idle_time: DBMetrics.db_client_connection_create_time(),
total_time: DBMetrics.db_client_operation_duration(),
queue_time: DBMetrics.db_client_connection_wait_time(),
query_time: DBMetrics.db_client_connection_use_time(),
]

defp add_measurements(attributes, measurements) do
Enum.reduce(@measurements, attributes, fn {telemetry_key, attribute_key}, attributes ->
case Map.get(measurements, telemetry_key) do
nil -> attributes
value -> Map.put(attributes, attribute_key, System.convert_time_unit(value, :native, :microsecond) / 1_000_000)
end
end)
end

defp maybe_add_db_system(attributes, Ecto.Adapters.Tds) do
Map.put(attributes, :"db.system", :mssql)
defp maybe_add_db_query_text(attributes, :enabled, query) do
Map.put(attributes, unquote(DBAttributes.db_query_text()), query)
end

defp maybe_add_db_system(attributes, _) do
defp maybe_add_db_query_text(attributes, :disabled, _query) do
attributes
end

defp maybe_add_db_query_text(attributes, sanitizer, query) when is_function(sanitizer, 1) do
Map.put(attributes, unquote(DBAttributes.db_query_text()), sanitizer.(query))
end

defp maybe_add_db_query_text(attributes, _, _query) do
attributes
end

defp add_additional_attributes(attributes, additional_attributes) do
Map.merge(attributes, additional_attributes)
end

# SHOULD be `{db.operation.name} {target}` if there is a (low-cardinality) {db.operation.name} available.
defp span_name(%{unquote(DBAttributes.db_operation_name()) => db_operation_name} = attributes), do: "#{db_operation_name} #{target(attributes)}"

# If there is no (low-cardinality) `db.operation.name` available, database span names SHOULD be `{target}`.
defp span_name(attributes), do: target(attributes)

# `db.collection.name` SHOULD be used for data manipulation operations or operations on database collections.
defp target(%{unquote(DBAttributes.db_collection_name()) => db_collection_name}), do: db_collection_name

# `db.namespace` SHOULD be used for operations on a specific database namespace.
defp target(%{unquote(DBAttributes.db_namespace()) => db_namespace}), do: db_namespace
end
9 changes: 5 additions & 4 deletions instrumentation/opentelemetry_ecto/mix.exs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,14 +57,15 @@ defmodule OpentelemetryEcto.MixProject do
defp deps do
[
{:telemetry, "~> 0.4 or ~> 1.0"},
{:opentelemetry_api, "~> 1.0"},
{:opentelemetry, "~> 1.0", only: [:dev, :test]},
{:opentelemetry_api, "~> 1.4"},
{:opentelemetry_process_propagator, "~> 0.3"},
{:opentelemetry_semantic_conventions, "~> 1.27"},
{:opentelemetry, "~> 1.5", only: [:dev, :test]},
{:opentelemetry_exporter, "~> 1.0", only: [:dev, :test]},
{:ex_doc, "~> 0.35", only: [:dev], runtime: false},
{:ecto_sql, ">= 3.0.0", only: [:dev, :test]},
{:postgrex, ">= 0.15.0", only: [:dev, :test]},
{:dialyxir, "~> 1.1", only: [:dev, :test], runtime: false},
{:opentelemetry_process_propagator, "~> 0.3"}
{:dialyxir, "~> 1.1", only: [:dev, :test], runtime: false}
]
end
end
7 changes: 4 additions & 3 deletions instrumentation/opentelemetry_ecto/mix.lock
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,12 @@
"makeup_elixir": {:hex, :makeup_elixir, "1.0.0", "74bb8348c9b3a51d5c589bf5aebb0466a84b33274150e3b6ece1da45584afc82", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "49159b7d7d999e836bedaf09dcf35ca18b312230cf901b725a64f3f42e407983"},
"makeup_erlang": {:hex, :makeup_erlang, "1.0.1", "c7f58c120b2b5aa5fd80d540a89fdf866ed42f1f3994e4fe189abebeab610839", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "8a89a1eeccc2d798d6ea15496a6e4870b75e014d1af514b1b71fa33134f57814"},
"nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"},
"opentelemetry": {:hex, :opentelemetry, "1.3.1", "f0a342a74379e3540a634e7047967733da4bc8b873ec9026e224b2bd7369b1fc", [:rebar3], [{:opentelemetry_api, "~> 1.2.2", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}], "hexpm", "de476b2ac4faad3e3fe3d6e18b35dec9cb338c3b9910c2ce9317836dacad3483"},
"opentelemetry_api": {:hex, :opentelemetry_api, "1.2.2", "693f47b0d8c76da2095fe858204cfd6350c27fe85d00e4b763deecc9588cf27a", [:mix, :rebar3], [{:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}], "hexpm", "dc77b9a00f137a858e60a852f14007bb66eda1ffbeb6c05d5fe6c9e678b05e9d"},
"opentelemetry": {:hex, :opentelemetry, "1.5.0", "7dda6551edfc3050ea4b0b40c0d2570423d6372b97e9c60793263ef62c53c3c2", [:rebar3], [{:opentelemetry_api, "~> 1.4", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "cdf4f51d17b592fc592b9a75f86a6f808c23044ba7cf7b9534debbcc5c23b0ee"},
"opentelemetry_api": {:hex, :opentelemetry_api, "1.4.0", "63ca1742f92f00059298f478048dfb826f4b20d49534493d6919a0db39b6db04", [:mix, :rebar3], [], "hexpm", "3dfbbfaa2c2ed3121c5c483162836c4f9027def469c41578af5ef32589fcfc58"},
"opentelemetry_exporter": {:hex, :opentelemetry_exporter, "1.6.0", "f4fbf69aa9f1541b253813221b82b48a9863bc1570d8ecc517bc510c0d1d3d8c", [:rebar3], [{:grpcbox, ">= 0.0.0", [hex: :grpcbox, repo: "hexpm", optional: false]}, {:opentelemetry, "~> 1.3", [hex: :opentelemetry, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.2", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:tls_certificate_check, "~> 1.18", [hex: :tls_certificate_check, repo: "hexpm", optional: false]}], "hexpm", "1802d1dca297e46f21e5832ecf843c451121e875f73f04db87355a6cb2ba1710"},
"opentelemetry_process_propagator": {:hex, :opentelemetry_process_propagator, "0.3.0", "ef5b2059403a1e2b2d2c65914e6962e56371570b8c3ab5323d7a8d3444fb7f84", [:mix, :rebar3], [{:opentelemetry_api, "~> 1.0", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "7243cb6de1523c473cba5b1aefa3f85e1ff8cc75d08f367104c1e11919c8c029"},
"opentelemetry_semantic_conventions": {:hex, :opentelemetry_semantic_conventions, "0.2.0", "b67fe459c2938fcab341cb0951c44860c62347c005ace1b50f8402576f241435", [:mix, :rebar3], [], "hexpm", "d61fa1f5639ee8668d74b527e6806e0503efc55a42db7b5f39939d84c07d6895"},
"opentelemetry_semantic_conventions": {:hex, :opentelemetry_semantic_conventions, "1.27.0", "acd0194a94a1e57d63da982ee9f4a9f88834ae0b31b0bd850815fe9be4bbb45f", [:mix, :rebar3], [], "hexpm", "9681ccaa24fd3d810b4461581717661fd85ff7019b082c2dff89c7d5b1fc2864"},
"opentelemetry_telemetry": {:hex, :opentelemetry_telemetry, "1.1.2", "410ab4d76b0921f42dbccbe5a7c831b8125282850be649ee1f70050d3961118a", [:mix, :rebar3], [{:opentelemetry_api, "~> 1.3", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.1", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "641ab469deb181957ac6d59bce6e1321d5fe2a56df444fc9c19afcad623ab253"},
"postgrex": {:hex, :postgrex, "0.17.4", "5777781f80f53b7c431a001c8dad83ee167bcebcf3a793e3906efff680ab62b3", [:mix], [{:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "6458f7d5b70652bc81c3ea759f91736c16a31be000f306d3c64bcdfe9a18b3cc"},
"ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"},
"telemetry": {:hex, :telemetry, "1.2.1", "68fdfe8d8f05a8428483a97d7aab2f268aaff24b49e0f599faa091f1d4e7f61c", [:rebar3], [], "hexpm", "dad9ce9d8effc621708f99eac538ef1cbe05d6a874dd741de2e689c47feafed5"},
Expand Down
Loading
Loading