otel-sqlparamstyle-processor/traces_processor_test.go at main · nilox94/otel-sqlparamstyle-processor · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
package sqlparamstyleprocessor

import (
	"context"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.opentelemetry.io/collector/component/componenttest"
	"go.opentelemetry.io/collector/consumer/consumertest"
	"go.opentelemetry.io/collector/pdata/ptrace"
	"go.opentelemetry.io/collector/processor/processortest"
)

func TestTracesProcessor_NormalizesPEP249ParamStyleInAttributes(t *testing.T) {
	cfg := &Config{
		AttributeKeys:     []string{"db.statement"},
		NormalizeInClause: true,
	}
	sink := new(consumertest.TracesSink)
	proc, err := newTracesProcessor(cfg, processortest.NewNopSettings(processortest.NopType), sink)
	require.NoError(t, err)

	err = proc.Start(context.Background(), componenttest.NewNopHost())
	require.NoError(t, err)
	t.Cleanup(func() { _ = proc.Shutdown(context.Background()) })

	td := ptrace.NewTraces()
	rs := td.ResourceSpans().AppendEmpty()
	ss := rs.ScopeSpans().AppendEmpty()
	span := ss.Spans().AppendEmpty()
	span.SetName("test-span")
	// PEP 249 format paramstyle
	span.Attributes().PutStr("db.statement", "SELECT * FROM x WHERE id IN (%s, %s, %s)")
	span.Attributes().PutStr("other", "unchanged")

	err = proc.ConsumeTraces(context.Background(), td)
	require.NoError(t, err)

	require.Len(t, sink.AllTraces(), 1)
	out := sink.AllTraces()[0]
	outSpan := out.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0)

	stmt, ok := outSpan.Attributes().Get("db.statement")
	require.True(t, ok)
	assert.NotEqual(t, "SELECT * FROM x WHERE id IN (%s, %s, %s)", stmt.Str(), "db.statement should be normalized")
	assert.NotEmpty(t, stmt.Str())

	other, ok := outSpan.Attributes().Get("other")
	require.True(t, ok)
	assert.Equal(t, "unchanged", other.Str())
}

func TestTracesProcessor_IgnoresNonConfiguredKeys(t *testing.T) {
	cfg := &Config{
		AttributeKeys:     []string{"db.statement"},
		NormalizeInClause: false,
	}
	sink := new(consumertest.TracesSink)
	proc, err := newTracesProcessor(cfg, processortest.NewNopSettings(processortest.NopType), sink)
	require.NoError(t, err)

	err = proc.Start(context.Background(), componenttest.NewNopHost())
	require.NoError(t, err)
	t.Cleanup(func() { _ = proc.Shutdown(context.Background()) })

	td := ptrace.NewTraces()
	rs := td.ResourceSpans().AppendEmpty()
	span := rs.ScopeSpans().AppendEmpty().Spans().AppendEmpty()
	span.Attributes().PutStr("custom.sql", "SELECT 1") // not in AttributeKeys

	err = proc.ConsumeTraces(context.Background(), td)
	require.NoError(t, err)

	out := sink.AllTraces()[0]
	outSpan := out.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0)
	custom, ok := outSpan.Attributes().Get("custom.sql")
	require.True(t, ok)
	assert.Equal(t, "SELECT 1", custom.Str(), "non-configured key should be left unchanged")
}

func TestTracesProcessor_InvalidSQLLeftUnchanged(t *testing.T) {
	cfg := &Config{
		AttributeKeys:     []string{"db.statement"},
		NormalizeInClause: true,
	}
	sink := new(consumertest.TracesSink)
	proc, err := newTracesProcessor(cfg, processortest.NewNopSettings(processortest.NopType), sink)
	require.NoError(t, err)

	err = proc.Start(context.Background(), componenttest.NewNopHost())
	require.NoError(t, err)
	t.Cleanup(func() { _ = proc.Shutdown(context.Background()) })

	td := ptrace.NewTraces()
	rs := td.ResourceSpans().AppendEmpty()
	span := rs.ScopeSpans().AppendEmpty().Spans().AppendEmpty()
	badSQL := "NOT VALID SQL {{{"
	span.Attributes().PutStr("db.statement", badSQL)

	err = proc.ConsumeTraces(context.Background(), td)
	require.NoError(t, err)

	out := sink.AllTraces()[0]
	outSpan := out.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0)
	stmt, ok := outSpan.Attributes().Get("db.statement")
	require.True(t, ok)
	assert.Equal(t, badSQL, stmt.Str(), "invalid SQL should be left unchanged")
}

func TestTracesProcessor_Capabilities(t *testing.T) {
	cfg := &Config{AttributeKeys: []string{"db.statement"}}
	sink := new(consumertest.TracesSink)
	proc, err := newTracesProcessor(cfg, processortest.NewNopSettings(processortest.NopType), sink)
	require.NoError(t, err)
	caps := proc.Capabilities()
	assert.True(t, caps.MutatesData)
}