1import { type HrTime, ValueType } from "@opentelemetry/api"
2import type * as Resources from "@opentelemetry/resources"
3import type {
4 CollectionResult,
5 DataPoint,
6 Histogram,
7 MetricCollectOptions,
8 MetricData,
9 MetricProducer
10} from "@opentelemetry/sdk-metrics"
11import { AggregationTemporality, DataPointType, InstrumentType } from "@opentelemetry/sdk-metrics"
12import type { InstrumentDescriptor } from "@opentelemetry/sdk-metrics/build/src/InstrumentDescriptor.js"
13import * as Arr from "effect/Array"
14import type * as Context from "effect/Context"
15import * as Metric from "effect/Metric"
16import * as Rec from "effect/Record"
17import type * as Metrics from "../OtelMetrics.ts"
18
19const sdkName = "@effect/opentelemetry/Metrics"
20
21type MetricDataWithInstrumentDescriptor = MetricData & {
22 readonly descriptor: InstrumentDescriptor
23}
24
25interface PreviousHistogramState {
26 readonly count: number
27 readonly sum: number
28 readonly bucketCounts: ReadonlyArray<number>
29 readonly min: number
30 readonly max: number
31}
32
33interface PreviousSummaryState {
34 readonly count: number
35 readonly sum: number
36}
37
38/** @internal */
39export class MetricProducerImpl implements MetricProducer {
40 resource: Resources.Resource
41 context: Context.Context<never>
42 temporality: Metrics.TemporalityPreference
43 startTimes: Map<string, HrTime>
44 startTimeNanos: HrTime
45 previousExportTimeNanos: HrTime
46 previousCounterState: Map<string, number | bigint>
47 previousHistogramState: Map<string, PreviousHistogramState>
48 previousFrequencyState: Map<string, Map<string, number>>
49 previousSummaryState: Map<string, PreviousSummaryState>
50
51 constructor(
52 resource: Resources.Resource,
53 context: Context.Context<never>,
54 temporality: Metrics.TemporalityPreference = "cumulative"
55 ) {
56 this.resource = resource
57 this.context = context
58 this.temporality = temporality
59 this.startTimes = new Map()
60 this.startTimeNanos = currentHrTime()
61 this.previousExportTimeNanos = this.startTimeNanos
62 this.previousCounterState = new Map()
63 this.previousHistogramState = new Map()
64 this.previousFrequencyState = new Map()
65 this.previousSummaryState = new Map()
66 }
67
68 fork(): MetricProducerImpl {
69 return new MetricProducerImpl(this.resource, this.context, this.temporality)
70 }
71
72 startTimeFor(name: string, hrTime: HrTime) {
73 if (this.startTimes.has(name)) {
74 return this.startTimes.get(name)!
75 }
76 this.startTimes.set(name, hrTime)
77 return hrTime
78 }
79
80 collect(_options?: MetricCollectOptions): Promise<CollectionResult> {
81 const snapshot = Metric.snapshotUnsafe(this.context)
82 const hrTimeNow = currentHrTime()
83 const metricData: Array<MetricData> = []
84 const metricDataByName = new Map<string, MetricData>()
85 const addMetricData = (data: MetricDataWithInstrumentDescriptor) => {
86 metricData.push(data)
87 metricDataByName.set(data.descriptor.name, data)
88 }
89
90 const isDelta = this.temporality === "delta"
91 const aggregationTemporality = isDelta
92 ? AggregationTemporality.DELTA
93 : AggregationTemporality.CUMULATIVE
94 const intervalStartTime = isDelta
95 ? this.previousExportTimeNanos
96 : this.startTimeNanos
97
98 for (let i = 0, len = snapshot.length; i < len; i++) {
99 const state = snapshot[i]
100 const attributes = state.attributes
101 ? Arr.reduce(Object.entries(state.attributes), {} as Record<string, string>, (acc, [key, value]) => {
102 Rec.assignProperty(acc, key, String(value))
103 return acc
104 })
105 : {}
106 const metricKey = makeMetricKey(state.id, state.attributes)
107
108 switch (state.type) {
109 case "Counter": {
110 const currentCount = state.state.count
111 let reportValue: number | bigint = currentCount
112
113 if (isDelta) {
114 const previousCount = this.previousCounterState.get(metricKey)
115 if (previousCount !== undefined) {
116 if (typeof currentCount === "bigint" && typeof previousCount === "bigint") {
117 reportValue = currentCount - previousCount
118 // Handle reset: if current < previous, report current value
119 if (state.state.incremental && reportValue < BigInt(0)) {
120 reportValue = currentCount
121 }
122 } else {
123 const curr = Number(currentCount)
124 const prev = Number(previousCount)
125 reportValue = curr - prev
126 // Handle reset
127 if (state.state.incremental && reportValue < 0) {
128 reportValue = curr
129 }
130 }
131 }
132 this.previousCounterState.set(metricKey, currentCount)
133 }
134
135 const descriptor = descriptorFromState(state, attributes)
136 const startTime = isDelta ? intervalStartTime : this.startTimeFor(descriptor.name, intervalStartTime)
137 const dataPoint: DataPoint<number> = {
138 startTime,
139 endTime: hrTimeNow,
140 attributes,
141 value: Number(reportValue)
142 }
143 if (metricDataByName.has(state.id)) {
144 metricDataByName.get(state.id)!.dataPoints.push(dataPoint as any)
145 } else {
146 addMetricData({
147 dataPointType: DataPointType.SUM,
148 descriptor,
149 isMonotonic: state.state.incremental,
150 aggregationTemporality,
151 dataPoints: [dataPoint]
152 })
153 }
154 break
155 }
156 case "Gauge": {
157 // Gauges don't have temporality - they always report current value
158 const descriptor = descriptorFromState(state, attributes)
159 const startTime = this.startTimeFor(descriptor.name, this.startTimeNanos)
160 const dataPoint: DataPoint<number> = {
161 startTime,
162 endTime: hrTimeNow,
163 attributes,
164 value: Number(state.state.value)
165 }
166 if (metricDataByName.has(state.id)) {
167 metricDataByName.get(state.id)!.dataPoints.push(dataPoint as any)
168 } else {
169 addMetricData({
170 dataPointType: DataPointType.GAUGE,
171 descriptor,
172 aggregationTemporality: AggregationTemporality.CUMULATIVE,
173 dataPoints: [dataPoint]
174 })
175 }
176 break
177 }
178 case "Histogram": {
179 const size = state.state.buckets.length
180 const currentBuckets = {
181 boundaries: Arr.allocate(size - 1) as Array<number>,
182 counts: Arr.allocate(size) as Array<number>
183 }
184 let idx = 0
185 let prev = 0
186 for (const [boundary, value] of state.state.buckets) {
187 if (idx < size - 1) {
188 currentBuckets.boundaries[idx] = boundary
189 }
190 currentBuckets.counts[idx] = value - prev
191 prev = value
192 idx++
193 }
194
195 let reportCount = state.state.count
196 let reportSum = state.state.sum
197 let reportBucketCounts = currentBuckets.counts
198 const reportMin = state.state.min
199 const reportMax = state.state.max
200
201 if (isDelta) {
202 const previousState = this.previousHistogramState.get(metricKey)
203 if (previousState !== undefined) {
204 reportCount = state.state.count - previousState.count
205 reportSum = state.state.sum - previousState.sum
206 reportBucketCounts = currentBuckets.counts.map((c, i) =>
207 Math.max(0, c - (previousState.bucketCounts[i] ?? 0))
208 )
209 }
210 this.previousHistogramState.set(metricKey, {
211 count: state.state.count,
212 sum: state.state.sum,
213 bucketCounts: currentBuckets.counts.slice(),
214 min: state.state.min,
215 max: state.state.max
216 })
217 }
218
219 const descriptor = descriptorFromState(state, attributes)
220 const startTime = isDelta ? intervalStartTime : this.startTimeFor(descriptor.name, intervalStartTime)
221 const dataPoint: DataPoint<Histogram> = {
222 startTime,
223 endTime: hrTimeNow,
224 attributes,
225 value: {
226 buckets: {
227 boundaries: currentBuckets.boundaries,
228 counts: reportBucketCounts
229 },
230 count: reportCount,
231 min: reportMin,
232 max: reportMax,
233 sum: reportSum
234 }
235 }
236
237 if (metricDataByName.has(state.id)) {
238 metricDataByName.get(state.id)!.dataPoints.push(dataPoint as any)
239 } else {
240 addMetricData({
241 dataPointType: DataPointType.HISTOGRAM,
242 descriptor,
243 aggregationTemporality,
244 dataPoints: [dataPoint]
245 })
246 }
247 break
248 }
249 case "Frequency": {
250 const dataPoints: Array<DataPoint<number>> = []
251 const currentOccurrences = new Map<string, number>()
252
253 for (const [freqKey, value] of state.state.occurrences) {
254 currentOccurrences.set(freqKey, value)
255 let reportValue = value
256
257 if (isDelta) {
258 const previousOccurrences = this.previousFrequencyState.get(metricKey)
259 if (previousOccurrences !== undefined) {
260 const previousValue = previousOccurrences.get(freqKey) ?? 0
261 reportValue = Math.max(0, value - previousValue)
262 }
263 }
264
265 const descriptor = descriptorFromState(state, attributes)
266 const startTime = isDelta ? intervalStartTime : this.startTimeFor(descriptor.name, intervalStartTime)
267 dataPoints.push({
268 startTime,
269 endTime: hrTimeNow,
270 attributes: {
271 ...attributes,
272 key: freqKey
273 },
274 value: reportValue
275 })
276 }
277
278 if (isDelta) {
279 this.previousFrequencyState.set(metricKey, currentOccurrences)
280 }
281
282 if (metricDataByName.has(state.id)) {
283 // oxlint-disable-next-line no-restricted-syntax
284 metricDataByName.get(state.id)!.dataPoints.push(...dataPoints as any)
285 } else {
286 const descriptor = descriptorFromState(state, attributes)
287 addMetricData({
288 dataPointType: DataPointType.SUM,
289 descriptor,
290 aggregationTemporality,
291 isMonotonic: true,
292 dataPoints
293 })
294 }
295 break
296 }
297 case "Summary": {
298 // Quantiles are always computed fresh from the sliding window
299 const dataPoints: Array<DataPoint<number>> = [{
300 startTime: intervalStartTime,
301 endTime: hrTimeNow,
302 attributes: { ...attributes, quantile: "min" },
303 value: state.state.min
304 }]
305 for (const [quantile, value] of state.state.quantiles) {
306 dataPoints.push({
307 startTime: intervalStartTime,
308 endTime: hrTimeNow,
309 attributes: { ...attributes, quantile: quantile.toString() },
310 value: value ?? 0
311 })
312 }
313 dataPoints.push({
314 startTime: intervalStartTime,
315 endTime: hrTimeNow,
316 attributes: { ...attributes, quantile: "max" },
317 value: state.state.max
318 })
319
320 let reportCount = state.state.count
321 let reportSum = state.state.sum
322
323 if (isDelta) {
324 const previousState = this.previousSummaryState.get(metricKey)
325 if (previousState !== undefined) {
326 reportCount = state.state.count - previousState.count
327 reportSum = state.state.sum - previousState.sum
328 }
329 this.previousSummaryState.set(metricKey, {
330 count: state.state.count,
331 sum: state.state.sum
332 })
333 }
334
335 const countDataPoint: DataPoint<number> = {
336 startTime: intervalStartTime,
337 endTime: hrTimeNow,
338 attributes,
339 value: reportCount
340 }
341 const sumDataPoint: DataPoint<number> = {
342 startTime: intervalStartTime,
343 endTime: hrTimeNow,
344 attributes,
345 value: reportSum
346 }
347
348 if (metricDataByName.has(`${state.id}_quantiles`)) {
349 // oxlint-disable-next-line no-restricted-syntax
350 metricDataByName.get(`${state.id}_quantiles`)!.dataPoints.push(...dataPoints as any)
351 metricDataByName.get(`${state.id}_count`)!.dataPoints.push(countDataPoint as any)
352 metricDataByName.get(`${state.id}_sum`)!.dataPoints.push(sumDataPoint as any)
353 } else {
354 const descriptor = descriptorFromState(state, attributes)
355 addMetricData({
356 dataPointType: DataPointType.SUM,
357 descriptor: {
358 ...descriptor,
359 name: `${descriptor.name}_quantiles`
360 },
361 aggregationTemporality,
362 isMonotonic: false,
363 dataPoints
364 })
365 addMetricData({
366 dataPointType: DataPointType.SUM,
367 descriptor: {
368 name: `${state.id}_count`,
369 description: state.description ?? "",
370 unit: "1",
371 type: InstrumentType.COUNTER,
372 valueType: ValueType.INT,
373 advice: {}
374 },
375 aggregationTemporality,
376 isMonotonic: true,
377 dataPoints: [countDataPoint]
378 })
379 addMetricData({
380 dataPointType: DataPointType.SUM,
381 descriptor: {
382 name: `${state.id}_sum`,
383 description: state.description ?? "",
384 unit: "1",
385 type: InstrumentType.COUNTER,
386 valueType: ValueType.DOUBLE,
387 advice: {}
388 },
389 aggregationTemporality,
390 isMonotonic: true,
391 dataPoints: [sumDataPoint]
392 })
393 }
394 break
395 }
396 }
397 }
398
399 // Update the previous export time for delta calculations
400 if (isDelta) {
401 this.previousExportTimeNanos = hrTimeNow
402 }
403
404 return Promise.resolve({
405 resourceMetrics: {
406 resource: this.resource,
407 scopeMetrics: [{
408 scope: { name: sdkName },
409 metrics: metricData
410 }]
411 },
412 errors: []
413 })
414 }
415}
416
417/** Creates a unique key for a metric including its attributes */
418const makeMetricKey = (id: string, attributes: Metric.Metric.AttributeSet | undefined): string => {
419 if (attributes === undefined || Object.keys(attributes).length === 0) {
420 return id
421 }
422 const sortedEntries = Object.entries(attributes).sort((a, b) => a[0].localeCompare(b[0]))
423 return `${id}:${JSON.stringify(sortedEntries)}`
424}
425
426const currentHrTime = (): HrTime => {
427 const now = Date.now()
428 return [Math.floor(now / 1000), (now % 1000) * 1000000]
429}
430
431const descriptorFromState = (
432 state: Metric.Metric.Snapshot,
433 attributes: Record<string, string>
434): InstrumentDescriptor => {
435 const unit = attributes.unit ?? attributes.time_unit ?? "1"
436 return {
437 name: state.id,
438 description: state.description ?? "",
439 unit,
440 type: instrumentTypeFromSnapshot(state),
441 valueType: determineValueType(state),
442 advice: {}
443 }
444}
445
446const instrumentTypeFromSnapshot = (state: Metric.Metric.Snapshot): InstrumentType => {
447 switch (state.type) {
448 case "Histogram":
449 return InstrumentType.HISTOGRAM
450 case "Gauge":
451 return InstrumentType.OBSERVABLE_GAUGE
452 case "Frequency":
453 return InstrumentType.COUNTER
454 case "Counter":
455 return state.state.incremental ? InstrumentType.COUNTER : InstrumentType.UP_DOWN_COUNTER
456 case "Summary":
457 return InstrumentType.COUNTER
458 }
459}
460
461const determineValueType = (state: Metric.Metric.Snapshot): ValueType => {
462 if (state.type === "Counter") {
463 return typeof state.state.count === "bigint" ? ValueType.INT : ValueType.DOUBLE
464 } else if (state.type === "Gauge") {
465 return typeof state.state.value === "bigint" ? ValueType.INT : ValueType.DOUBLE
466 }
467 return ValueType.DOUBLE
468}