Skip to content

usm, cnm: Report local container tags #37796

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

Draft
wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

usm, cnm: Report local container tags #37796

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
12 changes: 7 additions & 5 deletions cmd/system-probe/modules/network_tracer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"sync/atomic"
"time"

tagger "github.com/DataDog/datadog-agent/comp/core/tagger/def"
"github.com/DataDog/datadog-agent/pkg/network"
networkconfig "github.com/DataDog/datadog-agent/pkg/network/config"
"github.com/DataDog/datadog-agent/pkg/network/encoding/marshal"
Expand Down Expand Up @@ -57,14 +58,15 @@ func createNetworkTracerModule(_ *sysconfigtypes.Config, deps module.FactoryDepe

t, err := tracer.NewTracer(ncfg, deps.Telemetry, deps.Statsd)

return &networkTracer{tracer: t}, err
return &networkTracer{tracer: t, tagger: deps.Tagger}, err
}

var _ module.Module = &networkTracer{}

type networkTracer struct {
tracer *tracer.Tracer
restartTimer *time.Timer
tagger tagger.Component
}

func (nt *networkTracer) GetStats() map[string]interface{} {
Expand All @@ -88,7 +90,7 @@ func (nt *networkTracer) Register(httpMux *module.Router) error {
defer cleanup()
contentType := req.Header.Get("Accept")
marshaler := marshal.GetMarshaler(contentType)
writeConnections(w, marshaler, cs)
writeConnections(w, marshaler, cs, nt.tagger)

if nt.restartTimer != nil {
nt.restartTimer.Reset(inactivityRestartDuration)
Expand Down Expand Up @@ -129,7 +131,7 @@ func (nt *networkTracer) Register(httpMux *module.Router) error {

contentType := req.Header.Get("Accept")
marshaler := marshal.GetMarshaler(contentType)
writeConnections(w, marshaler, cs)
writeConnections(w, marshaler, cs, nt.tagger)
})

httpMux.HandleFunc("/debug/net_state", func(w http.ResponseWriter, req *http.Request) {
Expand Down Expand Up @@ -236,12 +238,12 @@ func logRequests(client string, count uint64, connectionsCount int, start time.T
}
}

func writeConnections(w http.ResponseWriter, marshaler marshal.Marshaler, cs *network.Connections) {
func writeConnections(w http.ResponseWriter, marshaler marshal.Marshaler, cs *network.Connections, tagger tagger.Component) {
defer network.Reclaim(cs)

w.Header().Set("Content-type", marshaler.ContentType())

connectionsModeler := marshal.NewConnectionsModeler(cs)
connectionsModeler := marshal.NewConnectionsModeler(cs, tagger)
defer connectionsModeler.Close()

err := marshaler.Marshal(cs, w, connectionsModeler)
Expand Down
4 changes: 2 additions & 2 deletions cmd/system-probe/modules/network_tracer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,13 +69,13 @@ func TestDecode(t *testing.T) {
marshaller := marshal.GetMarshaler(encoding.ContentTypeJSON)
ostream := bytes.NewBuffer(nil)

connectionsModeler := marshal.NewConnectionsModeler(in)
connectionsModeler := marshal.NewConnectionsModeler(in, nil)
defer connectionsModeler.Close()

err := marshaller.Marshal(in, ostream, connectionsModeler)
require.NoError(t, err)

writeConnections(rec, marshaller, in)
writeConnections(rec, marshaller, in, nil)

rec.Flush()
out := rec.Body.Bytes()
Expand Down
6 changes: 3 additions & 3 deletions pkg/network/encoding/encoding_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ func getBlobWriter(t *testing.T, assert *assert.Assertions, in *network.Connecti
marshaler := marshal.GetMarshaler(marshalerType)
assert.Equal(marshalerType, marshaler.ContentType())
blobWriter := bytes.NewBuffer(nil)
connectionsModeler := marshal.NewConnectionsModeler(in)
connectionsModeler := marshal.NewConnectionsModeler(in, nil)
defer connectionsModeler.Close()
err := marshaler.Marshal(in, blobWriter, connectionsModeler)
require.NoError(t, err)
Expand Down Expand Up @@ -377,7 +377,7 @@ func TestSerialization(t *testing.T) {
assert.Equal("application/json", marshaler.ContentType())

blobWriter := bytes.NewBuffer(nil)
connectionsModeler := marshal.NewConnectionsModeler(in)
connectionsModeler := marshal.NewConnectionsModeler(in, nil)
defer connectionsModeler.Close()
err := marshaler.Marshal(in, blobWriter, connectionsModeler)
require.NoError(t, err)
Expand Down Expand Up @@ -409,7 +409,7 @@ func TestSerialization(t *testing.T) {
assert.Equal("application/json", marshaler.ContentType())

blobWriter := bytes.NewBuffer(nil)
connectionsModeler := marshal.NewConnectionsModeler(in)
connectionsModeler := marshal.NewConnectionsModeler(in, nil)
defer connectionsModeler.Close()
err := marshaler.Marshal(in, blobWriter, connectionsModeler)
require.NoError(t, err)
Expand Down
17 changes: 16 additions & 1 deletion pkg/network/encoding/marshal/format.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,12 @@ import (
model "github.com/DataDog/agent-payload/v5/process"
"github.com/twmb/murmur3"

tagger "github.com/DataDog/datadog-agent/comp/core/tagger/def"
"github.com/DataDog/datadog-agent/comp/core/tagger/types"
"github.com/DataDog/datadog-agent/pkg/network"
"github.com/DataDog/datadog-agent/pkg/process/util"
"github.com/DataDog/datadog-agent/pkg/util/common"
"github.com/DataDog/datadog-agent/pkg/util/log"
)

const maxRoutes = math.MaxInt32
Expand Down Expand Up @@ -49,13 +53,23 @@ func mergeDynamicTags(dynamicTags ...map[string]struct{}) (out map[string]struct

// FormatConnection converts a ConnectionStats into an model.Connection
func FormatConnection(builder *model.ConnectionBuilder, conn network.ConnectionStats, routes map[network.Via]RouteIdx,
usmEncoders []usmEncoder, dnsFormatter *dnsFormatter, ipc ipCache, tagsSet *network.TagsSet) {
usmEncoders []usmEncoder, dnsFormatter *dnsFormatter, ipc ipCache, tagsSet *network.TagsSet, tagger tagger.Component) {

builder.SetPid(int32(conn.Pid))

var containerID string
var containerTags common.StringSet
if conn.ContainerID.Source != nil {
containerID = conn.ContainerID.Source.Get().(string)
if tagger != nil && containerID != "" {
entityID := types.NewEntityID(types.ContainerID, containerID)
entityTags, err := tagger.Tag(entityID, types.HighCardinality)
if err != nil {
log.Errorf("error getting container %s from metadata: %v", containerID, err)
} else {
containerTags = common.NewStringSet(entityTags...)
}
}
}
builder.SetLaddr(func(w *model.AddrBuilder) {
w.SetIp(ipc.Get(conn.Source))
Expand Down Expand Up @@ -123,6 +137,7 @@ func FormatConnection(builder *model.ConnectionBuilder, conn network.ConnectionS
dynamicTags = mergeDynamicTags(dynamicTags, encoderDynamicTags)
}

dynamicTags = mergeDynamicTags(dynamicTags, containerTags)
conn.StaticTags |= staticTags
tags, tagChecksum := formatTags(conn, tagsSet, dynamicTags)
for _, t := range tags {
Expand Down
7 changes: 5 additions & 2 deletions pkg/network/encoding/marshal/modeler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (

model "github.com/DataDog/agent-payload/v5/process"

tagger "github.com/DataDog/datadog-agent/comp/core/tagger/def"
pkgconfigsetup "github.com/DataDog/datadog-agent/pkg/config/setup"
"github.com/DataDog/datadog-agent/pkg/network"
)
Expand All @@ -26,21 +27,23 @@ type ConnectionsModeler struct {
ipc ipCache
routeIndex map[network.Via]RouteIdx
tagsSet *network.TagsSet
tagger tagger.Component
}

// NewConnectionsModeler initializes the connection modeler with encoders, dns formatter for
// the existing connections. The ConnectionsModeler holds the traffic encoders grouped by USM logic.
// It also includes formatted connection telemetry related to all batches, not specific batches.
// Furthermore, it stores the current agent configuration which applies to all instances related to the entire set of connections,
// rather than just individual batches.
func NewConnectionsModeler(conns *network.Connections) *ConnectionsModeler {
func NewConnectionsModeler(conns *network.Connections, tagger tagger.Component) *ConnectionsModeler {
ipc := make(ipCache, len(conns.Conns)/2)
return &ConnectionsModeler{
usmEncoders: initializeUSMEncoders(conns),
ipc: ipc,
dnsFormatter: newDNSFormatter(conns, ipc),
routeIndex: make(map[network.Via]RouteIdx),
tagsSet: network.NewTagsSet(),
tagger: tagger,
}
}

Expand All @@ -63,7 +66,7 @@ func (c *ConnectionsModeler) modelConnections(builder *model.ConnectionsBuilder,

for _, conn := range conns.Conns {
builder.AddConns(func(builder *model.ConnectionBuilder) {
FormatConnection(builder, conn, c.routeIndex, c.usmEncoders, c.dnsFormatter, c.ipc, c.tagsSet)
FormatConnection(builder, conn, c.routeIndex, c.usmEncoders, c.dnsFormatter, c.ipc, c.tagsSet, c.tagger)
})
}

Expand Down
2 changes: 1 addition & 1 deletion pkg/network/encoding/marshal/modeler_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ func TestConnectionModelerAgentConfiguration(t *testing.T) {
mock.NewSystemProbe(t)
cfgOnce = sync.Once{}
conns := &network.Connections{}
mod := NewConnectionsModeler(conns)
mod := NewConnectionsModeler(conns, nil)
streamer := NewProtoTestStreamer[*model.Connections]()
builder := model.NewConnectionsBuilder(streamer)
expected := &model.AgentConfiguration{
Expand Down
Loading