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.

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

Fix duplicate tags in TCP/UDP logs #29780

Merged
merged 11 commits into from
Oct 4, 2024
7 changes: 2 additions & 5 deletions pkg/logs/tailers/socket/tailer.go
Original file line number Diff line number Diff line change
Expand Up @@ -98,8 +98,7 @@ func (t *Tailer) readForever() {
log.Warnf("Couldn't read message from connection: %v", err)
return
}
copiedTags := make([]string, len(t.source.Config.Tags))
copy(copiedTags, t.source.Config.Tags)
msg := decoder.NewInput(data)
if ipAddress != "" && pkgconfigsetup.Datadog().GetBool("logs_config.use_sourcehost_tag") {
lastColonIndex := strings.LastIndex(ipAddress, ":")
var ipAddressWithoutPort string
Expand All @@ -109,10 +108,8 @@ func (t *Tailer) readForever() {
ipAddressWithoutPort = ipAddress
}
sourceHostTag := fmt.Sprintf("source_host:%s", ipAddressWithoutPort)
copiedTags = append(copiedTags, sourceHostTag)
msg.ParsingExtra.Tags = append(msg.ParsingExtra.Tags, sourceHostTag)
}
msg := decoder.NewInput(data)
msg.ParsingExtra.Tags = append(msg.ParsingExtra.Tags, copiedTags...)
t.decoder.InputChan <- msg
}
}
Expand Down
54 changes: 54 additions & 0 deletions pkg/logs/tailers/socket/tailer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"github.com/stretchr/testify/assert"

"github.com/DataDog/datadog-agent/comp/logs/agent/config"
pkgconfigsetup "github.com/DataDog/datadog-agent/pkg/config/setup"
"github.com/DataDog/datadog-agent/pkg/logs/message"
"github.com/DataDog/datadog-agent/pkg/logs/sources"
)
Expand Down Expand Up @@ -58,6 +59,49 @@ func TestReadShouldFailWithError(t *testing.T) {
tailer.Stop()
}

func TestSourceHostTag(t *testing.T) {
msgChan := make(chan *message.Message)
r, w := net.Pipe()
logsConfig := &config.LogsConfig{
Tags: []string{"test:tag"},
}

logSource := sources.NewLogSource("test-source", logsConfig)
tailer := NewTailer(logSource, r, msgChan, readWithIP)
tailer.Start()

var msg *message.Message
w.Write([]byte("foo\n"))
msg = <-msgChan
assert.Equal(t, []string{"source_host:192.168.1.100", "test:tag"}, msg.Tags())
tailer.Stop()
}

func TestSourceHostTagFlagDisabled(t *testing.T) {
// Set the config flag for source_host tag to false
pkgconfigsetup.Datadog().BindEnvAndSetDefault("logs_config.use_sourcehost_tag", false)

// Set up test components
msgChan := make(chan *message.Message)
r, w := net.Pipe()
logsConfig := &config.LogsConfig{
Tags: []string{"test:tag"},
}

logSource := sources.NewLogSource("test-source", logsConfig)
tailer := NewTailer(logSource, r, msgChan, readWithIP)
tailer.Start()

var msg *message.Message
w.Write([]byte("foo\n"))
msg = <-msgChan

// Assert that only the original tag is present (source_host tag should not be added)
assert.Equal(t, []string{"test:tag"}, msg.Tags(), "source_host tag should not be added when flag is disabled")

tailer.Stop()
}

func read(tailer *Tailer) ([]byte, string, error) {
inBuf := make([]byte, 4096)
n, err := tailer.Conn.Read(inBuf)
Expand All @@ -66,3 +110,13 @@ func read(tailer *Tailer) ([]byte, string, error) {
}
return inBuf[:n], "", nil
}

func readWithIP(tailer *Tailer) ([]byte, string, error) {
inBuf := make([]byte, 4096)
n, err := tailer.Conn.Read(inBuf)
if err != nil {
return nil, "", err
}
mockIPAddress := "192.168.1.100:8080"
return inBuf[:n], mockIPAddress, nil
}
12 changes: 12 additions & 0 deletions releasenotes/notes/fix-duplicate-tags-e97e8eeb6492235f.yaml
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
# Each section from every release note are combined when the
# CHANGELOG.rst is rendered. So the text needs to be worded so that
# it does not depend on any information only available in another
# section. This may mean repeating some details, but each section
# must be readable independently of the other.
#
# Each section note must be formatted as reStructuredText.
---
fixes:
- |
Fix duplicate tags in UDP/TCP logs.

Original file line number Diff line number Diff line change
Expand Up @@ -115,4 +115,5 @@ func assertLogsReceived(

t.Logf("stdout:\n\n%s\n\nstderr:\n\n%s", stdout, stderr)
utils.CheckLogsExpected(t, fakeIntake, "test-app", "bob", []string{sourceHostTag})
utils.CheckNoDuplicateTags(t, fakeIntake, "test-app", "bob")
}
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,32 @@ func CheckLogsExpected(t *testing.T, fakeIntake *components.FakeIntake, service,
}, 2*time.Minute, 10*time.Second)
}

// CheckNoDuplicateTags verifies that there is no duplicate tags
func CheckNoDuplicateTags(t *testing.T, fakeIntake *components.FakeIntake, service, content string) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it make sense to instead update the above CheckLogsExpected to check for duplicate tags?
That way we get dupe tag coverage for all test cases instead of just one. I don't think there is a good reason why we couldn't perform this check on all test cases.

t.Helper()

assert.EventuallyWithT(t, func(c *assert.CollectT) {
logs, err := FetchAndFilterLogs(fakeIntake, service, content)

if assert.NoErrorf(c, err, "Error fetching logs: %s", err) {
intakeLog := logsToString(logs)
if assert.NotEmpty(c, logs, "Expected logs with content: '%s' not found. Instead, found: %s", content, intakeLog) {
t.Logf("Logs from service: '%s' with content: '%s' collected", service, content)
log := logs[0]

// Use a map to check for duplicate tags
seenTags := make(map[string]struct{})
for _, tag := range log.Tags {
if _, exists := seenTags[tag]; exists {
t.Errorf("Duplicate tag found: %s", tag)
gh123man marked this conversation as resolved.
Show resolved Hide resolved
}
seenTags[tag] = struct{}{} // Mark the tag as seen
}
}
}
}, 2*time.Minute, 10*time.Second)
}

// CheckLogsNotExpected verifies the absence of unexpected logs.
func CheckLogsNotExpected(t *testing.T, fakeIntake *components.FakeIntake, service, content string) {
t.Helper()
Expand Down
Loading