Adding k8s remapper for supporting Kibana Inventory "Kubernetes Pod" page (#20)

* adding k8s remapper * adding event.module * Update remappers/k8smetrics/kubelet.go Co-authored-by: Tetiana Kravchenko <tanya.kravchenko.v@gmail.com> * Update remappers/k8smetrics/kubelet.go Co-authored-by: Tetiana Kravchenko <tanya.kravchenko.v@gmail.com> * merging into internal * fixing conflicts * merging with one internal metrics function * merging with one internal metrics function * adding tests * adding tests * fixing tests with correct results * fixing imports * fixing license lint error * adding dataset on top level * fixing tests with latest dataset changes * updating tests by moving common parts into internal folder * updating tests by moving common parts into internal folder * fixing field allighnement of structs linting errors --------- Co-authored-by: Tetiana Kravchenko <tanya.kravchenko.v@gmail.com>

Author: gizas

remappers/hostmetrics/hostmetrics_test.go

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package hostmetrics
+package internal
 
 import (
 "github.com/elastic/opentelemetry-lib/remappers/common"

remappers/internal/metric.go

@@ -0,0 +1,104 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package internal
+
+import (
+"testing"
+
+"go.opentelemetry.io/collector/pdata/pcommon"
+"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+type TestMetric struct {
+DP TestDP
+Name string
+Type pmetric.MetricType
+}
+
+type TestDP struct {
+Dbl *float64
+Int *int64
+Attrs map[string]any
+Ts pcommon.Timestamp
+}
+
+func MetricSliceToTestMetric(t *testing.T, ms pmetric.MetricSlice) []TestMetric {
+testMetrics := make([]TestMetric, ms.Len())
+for i := 0; i < ms.Len(); i++ {
+m := ms.At(i)
+testMetrics[i].Name = m.Name()
+testMetrics[i].Type = m.Type()
+
+var dps pmetric.NumberDataPointSlice
+switch m.Type() {
+case pmetric.MetricTypeGauge:
+dps = m.Gauge().DataPoints()
+case pmetric.MetricTypeSum:
+dps = m.Sum().DataPoints()
+}
+
+if dps.Len() != 1 {
+t.Fatalf("unexpected metric, test is written assuming each metric with a single datapoint")
+}
+
+dp := dps.At(0)
+testMetrics[i].DP = TestDP{Ts: dp.Timestamp(), Attrs: dp.Attributes().AsRaw()}
+switch dp.ValueType() {
+case pmetric.NumberDataPointValueTypeInt:
+testMetrics[i].DP.Int = Ptr(dp.IntValue())
+case pmetric.NumberDataPointValueTypeDouble:
+testMetrics[i].DP.Dbl = Ptr(dp.DoubleValue())
+}
+}
+
+return testMetrics
+}
+
+func TestMetricToMetricSlice(t testing.TB, testMetrics []TestMetric, out pmetric.MetricSlice) {
+out.EnsureCapacity(len(testMetrics))
+
+for _, testm := range testMetrics {
+m := out.AppendEmpty()
+m.SetName(testm.Name)
+
+var dps pmetric.NumberDataPointSlice
+switch typ := testm.Type; typ {
+case pmetric.MetricTypeGauge:
+dps = m.SetEmptyGauge().DataPoints()
+case pmetric.MetricTypeSum:
+dps = m.SetEmptySum().DataPoints()
+default:
+t.Fatalf("unhandled metric type %s", typ)
+}
+
+dp := dps.AppendEmpty()
+dp.SetTimestamp(testm.DP.Ts)
+if testm.DP.Int != nil {
+dp.SetIntValue(*testm.DP.Int)
+} else if testm.DP.Dbl != nil {
+dp.SetDoubleValue(*testm.DP.Dbl)
+}
+if err := dp.Attributes().FromRaw(testm.DP.Attrs); err != nil {
+t.Fatalf("failed to copy attributes from test data: %v", err)
+}
+}
+}
+
+func Ptr[T any](v T) *T {
+return &v
+}

remappers/internal/testing.go

@@ -0,0 +1,69 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package kubernetesmetrics
+
+import (
+remappers "github.com/elastic/opentelemetry-lib/remappers/internal"
+
+"go.opentelemetry.io/collector/pdata/pcommon"
+"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+func addClusterMetrics(
+src, out pmetric.MetricSlice,
+_ pcommon.Resource,
+dataset string,
+) error {
+var timestamp pcommon.Timestamp
+var node_allocatable_memory, node_allocatable_cpu int64
+
+// iterate all metrics in the current scope and generate the additional Elastic kubernetes integration metrics
+for i := 0; i < src.Len(); i++ {
+metric := src.At(i)
+if metric.Name() == "k8s.node.allocatable_cpu" {
+dp := metric.Gauge().DataPoints().At(0)
+if timestamp == 0 {
+timestamp = dp.Timestamp()
+}
+node_allocatable_cpu = dp.IntValue()
+} else if metric.Name() == "k8s.node.allocatable_memory" {
+dp := metric.Gauge().DataPoints().At(0)
+if timestamp == 0 {
+timestamp = dp.Timestamp()
+}
+node_allocatable_memory = dp.IntValue()
+}
+}
+
+remappers.AddMetrics(out, dataset, remappers.EmptyMutator,
+remappers.Metric{
+DataType: pmetric.MetricTypeGauge,
+Name: "kubernetes.node.cpu.allocatable.cores",
+Timestamp: timestamp,
+IntValue: &node_allocatable_cpu,
+},
+remappers.Metric{
+DataType: pmetric.MetricTypeGauge,
+Name: "kubernetes.node.memory.allocatable.bytes",
+Timestamp: timestamp,
+IntValue: &node_allocatable_memory,
+},
+)
+
+return nil
+}

remappers/kubernetesmetrics/clustermetrics.go

@@ -0,0 +1,41 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package kubernetesmetrics
+
+type config struct {
+KubernetesIntegrationDataset bool
+}
+
+// Option allows configuring the behavior of the kubernetes remapper.
+type Option func(config) config
+
+func newConfig(opts ...Option) (cfg config) {
+for _, opt := range opts {
+cfg = opt(cfg)
+}
+return cfg
+}
+
+// WithKubernetesIntegrationDataset sets the dataset of the remapped metrics as
+// as per the kubernetes integration. Example: kubernetes.pod
+func WithKubernetesIntegrationDataset(b bool) Option {
+return func(c config) config {
+c.KubernetesIntegrationDataset = b
+return c
+}
+}

remappers/kubernetesmetrics/config.go

@@ -0,0 +1,51 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package kubernetesmetrics
+
+import (
+"testing"
+
+"github.com/stretchr/testify/assert"
+)
+
+func TestConfig(t *testing.T) {
+for _, tc := range []struct {
+name string
+opts []Option
+expected config
+}{
+{
+name: "default",
+opts: nil,
+expected: config{
+KubernetesIntegrationDataset: false,
+},
+},
+{
+name: "k8s_integration_dataset",
+opts: []Option{WithKubernetesIntegrationDataset(true)},
+expected: config{
+KubernetesIntegrationDataset: true,
+},
+},
+} {
+t.Run(tc.name, func(t *testing.T) {
+assert.Equal(t, tc.expected, newConfig(tc.opts...))
+})
+}
+}

remappers/kubernetesmetrics/config_test.go

@@ -0,0 +1,104 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package kubernetesmetrics
+
+import (
+"path"
+"strings"
+
+"go.opentelemetry.io/collector/pdata/pcommon"
+"go.opentelemetry.io/collector/pdata/pmetric"
+"go.uber.org/zap"
+)
+
+var scraperToElasticDataset = map[string]string{
+"kubeletstatsreceiver": "kubernetes.pod",
+"k8sclusterreceiver": "kubernetes.node",
+}
+
+type remapFunc func(metrics pmetric.MetricSlice, out pmetric.MetricSlice, resource pcommon.Resource, dataset string) error
+
+// Remapper maps the OTel Kubernetes to Elastic Kubernetes metrics. These remapped
+// metrics power the curated Kibana dashboards. Each datapoint translated using
+// the remapper has the `event.processor` attribute set to `kubernetes`.
+type Remapper struct {
+logger *zap.Logger
+cfg config
+}
+
+// NewRemapper creates a new instance of kubernetes remapper.
+func NewRemapper(logger *zap.Logger, opts ...Option) *Remapper {
+return &Remapper{
+cfg: newConfig(opts...),
+logger: logger,
+}
+}
+
+var remapFuncs = map[string]remapFunc{
+"kubeletstatsreceiver": addKubeletMetrics,
+"k8sclusterreceiver": addClusterMetrics,
+}
+
+// Remap remaps an OTel ScopeMetrics to a list of OTel metrics such that the
+// remapped metrics could be trivially converted into Elastic system metrics.
+// It accepts the resource attributes to enrich the remapped metrics as per
+// Elastic convention. The current remapping logic assumes that each Metric
+// in the ScopeMetric will have datapoints for a single timestamp only. The
+// remapped metrics are added to the output `MetricSlice`.
+func (r *Remapper) Remap(
+src pmetric.ScopeMetrics,
+out pmetric.MetricSlice,
+resource pcommon.Resource,
+) {
+if !r.Valid(src) {
+return
+}
+
+scope := src.Scope()
+scraper := path.Base(scope.Name())
+
+var dataset string // an empty dataset defers setting dataset to the caller
+if r.cfg.KubernetesIntegrationDataset {
+var ok bool
+dataset, ok = scraperToElasticDataset[scraper]
+if !ok {
+r.logger.Warn("no dataset defined for scraper", zap.String("scraper", scraper))
+return
+}
+}
+
+remapFunc, ok := remapFuncs[scraper]
+if !ok {
+return
+}
+
+err := remapFunc(src.Metrics(), out, resource, dataset)
+if err != nil {
+r.logger.Warn(
+"failed to remap OTel kubernetes",
+zap.String("scope", scope.Name()),
+zap.Error(err),
+)
+}
+}
+
+// Valid validates a ScopeMetric against the kubernetes metrics remapper requirements.
+// Kubernetes remapper only remaps metrics from kubeletstatsreceiver or k8sclusterreceiver.
+func (r *Remapper) Valid(sm pmetric.ScopeMetrics) bool {
+return strings.HasPrefix(sm.Scope().Name(), "otelcol/kubeletstatsreceiver") || strings.HasPrefix(sm.Scope().Name(), "otelcol/k8sclusterreceiver")
+}