Add Container level metrics for CPU and memory resources. (amazon-con…

…tributing#150)

* Add container level CPU and memory metrics collection

    1. Add metric collection at container level
    2. Refactor existing kubelet code to make it unit testable.
    3. Add units for kubelet to test pod, node and container level metric collection.

* Refactor: rename port -> hostPort and summaryProvider -> kubeletSummaryProvider

* Refactor: Change naming in kubelet

2. Remove pas by reference to extractors

* Refactored: address naming inconsistencies

* Refactor: remove extra GetClient

* Fix merge conflict

receiver/awscontainerinsightreceiver/internal/k8swindows/extractors/extractor.go

@@ -36,7 +36,7 @@ type RawMetric struct {
 }
 
 type MetricExtractor interface {
-	HasValue(summary *RawMetric) bool
-	GetValue(summary *RawMetric, mInfo cExtractor.CPUMemInfoProvider, containerType string) []*cExtractor.CAdvisorMetric
+	HasValue(summary RawMetric) bool
+	GetValue(summary RawMetric, mInfo cExtractor.CPUMemInfoProvider, containerType string) []*cExtractor.CAdvisorMetric
 	Shutdown() error
 }

receiver/awscontainerinsightreceiver/internal/k8swindows/k8swindows.go

@@ -15,19 +15,20 @@ import (
 	cExtractor "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/cadvisor/extractors"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/host"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/k8swindows/extractors"
+	kubeletsummaryprovider "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/k8swindows/kubelet"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/stores"
 
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.uber.org/zap"
 )
 
 type K8sWindows struct {
-	cancel          context.CancelFunc
-	logger          *zap.Logger
-	nodeName        string `toml:"node_name"`
-	k8sDecorator    stores.K8sDecorator
-	summaryProvider *kubeletSummaryProvider
-	hostInfo        host.Info
+	cancel                 context.CancelFunc
+	logger                 *zap.Logger
+	nodeName               string `toml:"node_name"`
+	k8sDecorator           stores.K8sDecorator
+	kubeletSummaryProvider *kubeletsummaryprovider.SummaryProvider
+	hostInfo               host.Info
 }
 
 var metricsExtractors = []extractors.MetricExtractor{}
@@ -37,36 +38,39 @@ func New(logger *zap.Logger, decorator *stores.K8sDecorator, hostInfo host.Info)
 	if nodeName == "" {
 		return nil, errors.New("missing environment variable HOST_NAME. Please check your deployment YAML config")
 	}
-	k8sSummaryProvider, err := new(logger, hostInfo)
-	if err != nil {
-		logger.Error("failed to initialize kubelet summary provider, ", zap.Error(err))
-		return nil, err
-	}
 
 	metricsExtractors = []extractors.MetricExtractor{}
 	metricsExtractors = append(metricsExtractors, extractors.NewCPUMetricExtractor(logger))
 	metricsExtractors = append(metricsExtractors, extractors.NewMemMetricExtractor(logger))
+
+	ksp, err := kubeletsummaryprovider.New(logger, &hostInfo, metricsExtractors)
+	if err != nil {
+		logger.Error("failed to initialize kubelet SummaryProvider, ", zap.Error(err))
+		return nil, err
+	}
+
 	return &K8sWindows{
-		logger:          logger,
-		nodeName:        nodeName,
-		k8sDecorator:    *decorator,
-		summaryProvider: k8sSummaryProvider,
-		hostInfo:        hostInfo,
+		logger:                 logger,
+		nodeName:               nodeName,
+		k8sDecorator:           *decorator,
+		kubeletSummaryProvider: ksp,
+		hostInfo:               hostInfo,
 	}, nil
 }
 
 func (k *K8sWindows) GetMetrics() []pmetric.Metrics {
 	k.logger.Debug("D! called K8sWindows GetMetrics")
 	var result []pmetric.Metrics
 
-	metrics, err := k.summaryProvider.getMetrics()
+	metrics, err := k.kubeletSummaryProvider.GetMetrics()
 	if err != nil {
-		k.logger.Error("error getting metrics from kubelet summary provider, ", zap.Error(err))
+		k.logger.Error("failed to get metrics from kubelet SummaryProvider, ", zap.Error(err))
 		return result
 	}
+	metrics = cExtractor.MergeMetrics(metrics)
 	metrics = k.decorateMetrics(metrics)
-	for _, k8sSummaryMetric := range metrics {
-		md := ci.ConvertToOTLPMetrics(k8sSummaryMetric.GetFields(), k8sSummaryMetric.GetTags(), k.logger)
+	for _, ciMetric := range metrics {
+		md := ci.ConvertToOTLPMetrics(ciMetric.GetFields(), ciMetric.GetTags(), k.logger)
 		result = append(result, md)
 	}
 
@@ -116,7 +120,3 @@ func (k *K8sWindows) Shutdown() error {
 	k.logger.Debug("D! called K8sWindows Shutdown")
 	return nil
 }
-
-func GetMetricsExtractors() []extractors.MetricExtractor {
-	return metricsExtractors
-}

receiver/awscontainerinsightreceiver/internal/k8swindows/kubelet/client.go

@@ -0,0 +1,52 @@
+//go:build windows
+// +build windows
+
+package kubelet
+
+import (
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/stores/kubeletutil"
+	"go.uber.org/zap"
+	stats "k8s.io/kubelet/pkg/apis/stats/v1alpha1"
+)
+
+// KubeletProvider Represents interface to kubelet.
+type KubeletProvider interface {
+	GetSummary() (*stats.Summary, error)
+}
+
+type kubeletProvider struct {
+	logger   *zap.Logger
+	hostIP   string
+	hostPort string
+	client   *kubeletutil.KubeletClient
+}
+
+// getClient Returns singleton kubelet client.
+func (kp *kubeletProvider) getClient() (*kubeletutil.KubeletClient, error) {
+	if kp.client != nil {
+		return kp.client, nil
+	}
+	kclient, err := kubeletutil.NewKubeletClient(kp.hostIP, kp.hostPort, kp.logger)
+	if err != nil {
+		kp.logger.Error("failed to initialize new kubelet client, ", zap.Error(err))
+		return nil, err
+	}
+	kp.client = kclient
+	return kclient, nil
+}
+
+// GetSummary Get Summary from kubelet API.
+func (kp *kubeletProvider) GetSummary() (*stats.Summary, error) {
+	kclient, err := kp.getClient()
+	if err != nil {
+		kp.logger.Error("failed to get kubelet client, ", zap.Error(err))
+		return nil, err
+	}
+
+	summary, err := kclient.Summary(kp.logger)
+	if err != nil {
+		kp.logger.Error("failure from kubelet on getting summary, ", zap.Error(err))
+		return nil, err
+	}
+	return summary, nil
+}