Skip to content

Commit bee7cb1

Browse files
pmatyjasek-sumopmatyjasek-sumo
authored
[k8sprocessor] Add ability to associate metadata tags using pod UID rather than just IP (#2199)
Add ability to associate metadata tags in k8s processor using pod UID rather than just IP **Link to tracking Issue:** #1146 **Testing:** Unit tests **Documentation:** In-code comments
1 parent 57c9d04 commit bee7cb1

File tree

16 files changed

+638
-183
lines changed

16 files changed

+638
-183
lines changed

‎processor/k8sprocessor/client_test.go‎

Lines changed: 17 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -27,11 +27,12 @@ import (
2727

2828
// fakeClient is used as a replacement for WatchClient in test cases.
2929
type fakeClient struct {
30-
Pods map[string]*kube.Pod
31-
Rules kube.ExtractionRules
32-
Filters kube.Filters
33-
Informer cache.SharedInformer
34-
StopCh chan struct{}
30+
Pods map[kube.PodIdentifier]*kube.Pod
31+
Rules kube.ExtractionRules
32+
Filters kube.Filters
33+
Associations []kube.Association
34+
Informer cache.SharedInformer
35+
StopCh chan struct{}
3536
}
3637

3738
func selectors() (labels.Selector, fields.Selector) {
@@ -40,22 +41,24 @@ func selectors() (labels.Selector, fields.Selector) {
4041
}
4142

4243
// newFakeClient instantiates a new FakeClient object and satisfies the ClientProvider type
43-
func newFakeClient(_ *zap.Logger, apiCfg k8sconfig.APIConfig, rules kube.ExtractionRules, filters kube.Filters, _ kube.APIClientsetProvider, _ kube.InformerProvider) (kube.Client, error) {
44+
func newFakeClient(_ *zap.Logger, apiCfg k8sconfig.APIConfig, rules kube.ExtractionRules, filters kube.Filters, associations []kube.Association, _ kube.APIClientsetProvider, _ kube.InformerProvider) (kube.Client, error) {
4445
cs := fake.NewSimpleClientset()
4546

4647
ls, fs := selectors()
4748
return &fakeClient{
48-
Pods: map[string]*kube.Pod{},
49-
Rules: rules,
50-
Filters: filters,
51-
Informer: kube.NewFakeInformer(cs, "", ls, fs),
52-
StopCh: make(chan struct{}),
49+
Pods: map[kube.PodIdentifier]*kube.Pod{},
50+
Rules: rules,
51+
Filters: filters,
52+
Associations: associations,
53+
Informer: kube.NewFakeInformer(cs, "", ls, fs),
54+
StopCh: make(chan struct{}),
5355
}, nil
5456
}
5557

56-
// GetPodByIP looks up FakeClient.Pods map by the provided string.
57-
func (f *fakeClient) GetPodByIP(ip string) (*kube.Pod, bool) {
58-
p, ok := f.Pods[ip]
58+
// GetPod looks up FakeClient.Pods map by the provided string,
59+
// which might represent either IP address or Pod UID.
60+
func (f *fakeClient) GetPod(identifier kube.PodIdentifier) (*kube.Pod, bool) {
61+
p, ok := f.Pods[identifier]
5962
return p, ok
6063
}
6164

‎processor/k8sprocessor/config.go‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,10 @@ type Config struct {
3939
// Filter section allows specifying filters to filter
4040
// pods by labels, fields, namespaces, nodes, etc.
4141
Filter FilterConfig `mapstructure:"filter"`
42+
43+
// Association section allows to define rules for tagging spans, metrics,
44+
// and logs with Pod metadata.
45+
Association []PodAssociationConfig `mapstructure:"pod_association"`
4246
}
4347

4448
// ExtractConfig section allows specifying extraction rules to extract
@@ -173,3 +177,15 @@ type FieldFilterConfig struct {
173177
// equals, not-equals, exists, does-not-exist.
174178
Op string `mapstructure:"op"`
175179
}
180+
181+
// PodAssociationConfig contain single rule how to associate Pod metadata
182+
// with logs, spans and metrics
183+
type PodAssociationConfig struct {
184+
// From represents the source of the association.
185+
// Allowed values are "connection" and "labels".
186+
From string `mapstructure:"from"`
187+
188+
// Name represents extracted key name.
189+
// e.g. ip, pod_uid, k8s.pod.ip
190+
Name string `mapstructure:"name"`
191+
}

‎processor/k8sprocessor/config_test.go‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,5 +89,27 @@ func TestLoadConfig(t *testing.T) {
8989
{Key: "key2", Value: "value2", Op: "not-equals"},
9090
},
9191
},
92+
Association: []PodAssociationConfig{
93+
{
94+
From: "resource_attribute",
95+
Name: "ip",
96+
},
97+
{
98+
From: "resource_attribute",
99+
Name: "k8s.pod.ip",
100+
},
101+
{
102+
From: "resource_attribute",
103+
Name: "host.name",
104+
},
105+
{
106+
From: "connection",
107+
Name: "ip",
108+
},
109+
{
110+
From: "resource_attribute",
111+
Name: "k8s.pod.uid",
112+
},
113+
},
92114
})
93115
}

‎processor/k8sprocessor/doc.go‎

Lines changed: 24 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,30 @@
1515
// Package k8sprocessor allow automatic tagging of spans, metrics and logs with k8s metadata.
1616
//
1717
// The processor automatically discovers k8s resources (pods), extracts metadata from them and adds the
18-
// extracted metadata to the relevant spans, metrics and logs. The processor use the kubernetes API to discover all pods
19-
// running in a cluster, keeps a record of their IP addresses and interesting metadata. Upon receiving telemetry data,
20-
// the processor looks for presence of well-known resource attributes which might contain IP address ("ip",
21-
// "k8s.pod.ip" for logs, metrics or traces and "host.name" for metrics). If this field is not available, or it
22-
// does not contain a valid IP address, the processor tries to identify the source IP address of the service
23-
// that sent the telemetry data.
24-
// If a match is found, the cached metadata is added to the data as resource attributes.
18+
// extracted metadata to the relevant spans, metrics and logs. The processor uses the kubernetes API to discover all pods
19+
// running in a cluster, keeps a record of their IP addresses, pod UIDs and interesting metadata.
20+
// The rules for associating the data passing through the processor (spans, metrics and logs) with specific Pod Metadata are configured via "pod_association" key.
21+
// It represents a list of rules that are executed in the specified order until the first one is able to do the match.
22+
// Each rule is specified as a pair of from (representing the rule type) and name (representing the extracted key name).
23+
// Following rule types are available:
24+
// from: "resource_attribute" - allows to specify the attribute name to lookup up in the list of attributes of the received Resource. The specified attribute, if it is present, identifies the Pod that is represented by the Resource.
25+
// (the value can contain either IP address or Pod UID)
26+
// from: "connection" - takes the IP attribute from connection context (if available) and automatically
27+
// associates it with "k8s.pod.ip" attribute
28+
// Pod association configuration.
29+
// pod_association:
30+
// - from: resource_attribute
31+
// name: ip
32+
// - from: resource_attribute
33+
// name: k8s.pod.ip
34+
// - from: resource_attribute
35+
// name: host.name
36+
// - from: connection
37+
// name: ip
38+
// - from: resource_attribute
39+
// name: k8s.pod.uid
40+
//
41+
// If Pod association rules are not configured resources are associated with metadata only by connection's IP Address.
2542
//
2643
// RBAC
2744
//

‎processor/k8sprocessor/factory.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -190,5 +190,7 @@ func createProcessorOpts(cfg configmodels.Processor) []Option {
190190
opts = append(opts, WithFilterFields(oCfg.Filter.Fields...))
191191
opts = append(opts, WithAPIConfig(oCfg.APIConfig))
192192

193+
opts = append(opts, WithExtractPodAssociations(oCfg.Association...))
194+
193195
return opts
194196
}

‎processor/k8sprocessor/ip_extractor.go‎

Lines changed: 0 additions & 58 deletions
This file was deleted.

0 commit comments

Comments
 (0)