Merge pull request #346 from bobrik/ivan/decoder-cache-partition

Partition decoder cache by name

decoder/decoder.go

@@ -23,7 +23,7 @@ type Decoder interface {
 type Set struct {
 	mu       sync.Mutex
 	decoders map[string]Decoder
-	cache    map[string][]string
+	cache    map[string]map[string][]string
 }
 
 // NewSet creates a Set with all known decoders
@@ -57,7 +57,7 @@ func NewSet() (*Set, error) {
 			"syscall":      &Syscall{},
 			"uint":         &UInt{},
 		},
-		cache: map[string][]string{},
+		cache: map[string]map[string][]string{},
 	}, nil
 }
 
@@ -85,14 +85,20 @@ func (s *Set) decode(in []byte, label config.Label) ([]byte, error) {
 }
 
 // DecodeLabels transforms eBPF map key bytes into a list of label values
-// according to configuration
-func (s *Set) DecodeLabels(in []byte, labels []config.Label) ([]string, error) {
+// according to configuration (different label sets require different names)
+func (s *Set) DecodeLabels(in []byte, name string, labels []config.Label) ([]string, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	cache, ok := s.cache[name]
+	if !ok {
+		cache = map[string][]string{}
+		s.cache[name] = cache
+	}
+
 	// string(in) must not be a variable to avoid allocation:
 	// * https://github.com/golang/go/commit/f5f5a8b6209f8
-	if cached, ok := s.cache[string(in)]; ok {
+	if cached, ok := cache[string(in)]; ok {
 		return cached, nil
 	}
 
@@ -101,7 +107,7 @@ func (s *Set) DecodeLabels(in []byte, labels []config.Label) ([]string, error) {
 		return nil, err
 	}
 
-	s.cache[string(in)] = values
+	cache[string(in)] = values
 
 	return values, nil
 }

decoder/decoder_test.go

@@ -121,13 +121,13 @@ func TestDecodeLabels(t *testing.T) {
 		},
 	}
 
-	for _, c := range cases {
+	for i, c := range cases {
 		s, err := NewSet()
 		if err != nil {
 			t.Fatal(err)
 		}
 
-		out, err := s.DecodeLabels(c.in, c.labels)
+		out, err := s.DecodeLabels(c.in, fmt.Sprintf("test:%d", i), c.labels)
 		if c.err {
 			if err == nil {
 				t.Errorf("Expected error for input %#v and labels %#v, but did not receive it", c.in, c.labels)
@@ -152,7 +152,7 @@ func TestDecodeLabels(t *testing.T) {
 	}
 }
 
-func TestConcurrency(t *testing.T) {
+func TestDecoderSetConcurrency(t *testing.T) {
 	in := append([]byte{0x8, 0x0, 0x0, 0x0}, zeroPaddedString("bananas", 32)...)
 
 	labels := []config.Label{
@@ -197,7 +197,7 @@ func TestConcurrency(t *testing.T) {
 		go func() {
 			defer wg.Done()
 
-			_, err := s.DecodeLabels(in, labels)
+			_, err := s.DecodeLabels(in, "concurrency", labels)
 			if err != nil {
 				t.Error(err)
 			}
@@ -207,6 +207,66 @@ func TestConcurrency(t *testing.T) {
 	wg.Wait()
 }
 
+func TestDecoderSetCache(t *testing.T) {
+	in := []byte{0xba, 0xbe, 0xba, 0xbe, 0xde, 0xad, 0xbe, 0xef}
+
+	one := []config.Label{
+		{
+			Name: "single_u64",
+			Size: 8,
+			Decoders: []config.Decoder{
+				{
+					Name: "uint",
+				},
+			},
+		},
+	}
+
+	two := []config.Label{
+		{
+			Name: "u32_one",
+			Size: 4,
+			Decoders: []config.Decoder{
+				{
+					Name: "uint",
+				},
+			},
+		},
+		{
+			Name: "u32_two",
+			Size: 4,
+			Decoders: []config.Decoder{
+				{
+					Name: "uint",
+				},
+			},
+		},
+	}
+
+	s, err := NewSet()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	single, err := s.DecodeLabels(in, "one", one)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(single) != 1 {
+		t.Errorf("Expected one u64 from %#v, got %#v", one, single)
+	}
+
+	double, err := s.DecodeLabels(in, "two", two)
+	if err != nil {
+		t.Error(err)
+	}
+
+	if len(double) != 2 {
+		t.Errorf("Expected two u32 from %#v, got %#v", two, double)
+	}
+}
+
 func BenchmarkCache(b *testing.B) {
 	in := []byte{
 		0x8, 0xab, 0xce, 0xef,
@@ -270,7 +330,7 @@ func BenchmarkCache(b *testing.B) {
 
 	b.Run("cached", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
-			_, err := s.DecodeLabels(in, labels)
+			_, err := s.DecodeLabels(in, "test", labels)
 			if err != nil {
 				b.Fatal(err)
 			}

exporter/exporter.go

@@ -438,7 +438,7 @@ func (e *Exporter) mapValues(module *libbpfgo.Module, name string, labels []conf
 			raw = raw[4:]
 		}
 
-		metricValues[i].labels, err = e.decoders.DecodeLabels(raw, labels)
+		metricValues[i].labels, err = e.decoders.DecodeLabels(raw, name, labels)
 		if err != nil {
 			if err == decoder.ErrSkipLabelSet {
 				continue

exporter/perf_event_array.go

@@ -43,7 +43,7 @@ func newPerfEventArraySink(decoders *decoder.Set, module *libbpfgo.Module, count
 				validDataSize += labelConfig.Size
 			}
 
-			labelValues, err := decoders.DecodeLabels(rawBytes[:validDataSize], sink.counterConfig.Labels)
+			labelValues, err := decoders.DecodeLabels(rawBytes[:validDataSize], counterConfig.Name, sink.counterConfig.Labels)
 			if err != nil {
 				if err == decoder.ErrSkipLabelSet {
 					continue