[receiver/hostmetrics] add EntityState entity event as log record for…

… host

.chloggen/entity-event.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: hostmetricsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "allow configuring log pipeline to send host EntityState event"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33927]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/datadogexporter/go.mod

@@ -253,6 +253,8 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/docker v0.104.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter v0.104.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/pdatautil v0.104.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.104.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata v0.104.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.104.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.104.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/sampling v0.104.0 // indirect

receiver/hostmetricsreceiver/README.md

@@ -3,19 +3,21 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [beta]: metrics   |
+| Stability     | [development]: logs   |
+|               | [beta]: metrics   |
 | Distributions | [core], [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Fhostmetrics%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Fhostmetrics) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Fhostmetrics%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Fhostmetrics) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@dmitryax](https://www.github.com/dmitryax), [@braydonk](https://www.github.com/braydonk) |
 
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
 [beta]: https://github.com/open-telemetry/opentelemetry-collector#beta
 [core]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 <!-- end autogenerated section -->
 
 The Host Metrics receiver generates metrics about the host system scraped
-from various sources. This is intended to be used when the collector is
-deployed as an agent.
+from various sources and host entity event as log. This is intended to be 
+used when the collector is deployed as an agent.
 
 ## Getting Started
 
@@ -188,6 +190,9 @@ Currently, the hostmetrics receiver does not set any Resource attributes on the
 ```
 export OTEL_RESOURCE_ATTRIBUTES="service.name=<the name of your service>,service.namespace=<the namespace of your service>,service.instance.id=<uuid of the instance>"
 ```
+## Entity Events
+
+The hostmetrics receiver can send entity event as log. See the [Entites Data Model](https://github.com/open-telemetry/oteps/blob/main/text/entities/0256-entities-data-model.md#entity-events) spec for more details. You can configure hostmetrics receiver to send periodic EntityState event by setting `metadata_collection_interval` (default = 0).
 
 ## Feature Gates
 

receiver/hostmetricsreceiver/config.go

@@ -6,6 +6,7 @@ package hostmetricsreceiver // import "github.com/open-telemetry/opentelemetry-c
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap"
@@ -25,6 +26,13 @@ type Config struct {
 	Scrapers                       map[string]internal.Config `mapstructure:"-"`
 	// RootPath is the host's root directory (linux only).
 	RootPath string `mapstructure:"root_path"`
+
+	// Collection interval for metadata.
+	// Metadata of the particular entity in the cluster is collected when the entity changes.
+	// In addition metadata of all entities is collected periodically even if no changes happen.
+	// Setting the duration to 0 will disable periodic collection (however will not impact
+	// metadata collection on changes).
+	MetadataCollectionInterval time.Duration `mapstructure:"metadata_collection_interval"`
 }
 
 var _ component.Config = (*Config)(nil)

receiver/hostmetricsreceiver/factory.go

@@ -48,7 +48,8 @@ func NewFactory() receiver.Factory {
 	return receiver.NewFactory(
 		metadata.Type,
 		createDefaultConfig,
-		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability))
+		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability),
+		receiver.WithLogs(createLogsReceiver, metadata.MetricsStability))
 }
 
 func getScraperFactory(key string) (internal.ScraperFactory, bool) {
@@ -71,8 +72,20 @@ func createMetricsReceiver(
 	cfg component.Config,
 	consumer consumer.Metrics,
 ) (receiver.Metrics, error) {
+	var err error
 	oCfg := cfg.(*Config)
 
+	r := receivers.GetOrAdd(
+		cfg, func() component.Component {
+			var rcv component.Component
+			rcv, err = newHostMetricsReceiver(oCfg, &set)
+			return rcv
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+
 	addScraperOptions, err := createAddScraperOptions(ctx, set, oCfg, scraperFactories)
 	if err != nil {
 		return nil, err
@@ -81,14 +94,39 @@ func createMetricsReceiver(
 	host.EnableBootTimeCache(true)
 	process.EnableBootTimeCache(true)
 
-	return scraperhelper.NewScraperControllerReceiver(
+	scraperHelper, err := scraperhelper.NewScraperControllerReceiver(
 		&oCfg.ControllerConfig,
 		set,
 		consumer,
 		addScraperOptions...,
 	)
+	if err != nil {
+		return nil, err
+	}
+
+	r.Unwrap().(*hostMetricsReceiver).scraperHelper = scraperHelper
+	return r, nil
 }
 
+func createLogsReceiver(
+	ctx context.Context, set receiver.Settings, cfg component.Config, consumer consumer.Logs,
+) (receiver.Logs, error) {
+	var err error
+	oCfg := cfg.(*Config)
+
+	r := receivers.GetOrAdd(
+		cfg, func() component.Component {
+			var rcv component.Component
+			rcv, err = newHostMetricsReceiver(oCfg, &set)
+			return rcv
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+	r.Unwrap().(*hostMetricsReceiver).nextLogs = consumer
+	return r, nil
+}
 func createAddScraperOptions(
 	ctx context.Context,
 	set receiver.Settings,

receiver/hostmetricsreceiver/factory_test.go

@@ -39,8 +39,8 @@ func TestCreateReceiver(t *testing.T) {
 	assert.NotNil(t, mReceiver)
 
 	tLogs, err := factory.CreateLogsReceiver(context.Background(), creationSet, cfg, consumertest.NewNop())
-	assert.Equal(t, err, component.ErrDataTypeIsNotSupported)
-	assert.Nil(t, tLogs)
+	assert.NoError(t, err)
+	assert.NotNil(t, tLogs)
 }
 
 func TestCreateReceiver_ScraperKeyConfigError(t *testing.T) {

receiver/hostmetricsreceiver/go.mod

@@ -6,6 +6,8 @@ require (
 	github.com/google/go-cmp v0.6.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.104.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter v0.104.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.104.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata v0.104.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.104.0
 	github.com/prometheus-community/windows_exporter v0.25.1
 	github.com/prometheus/procfs v0.15.1

receiver/hostmetricsreceiver/metadata.yaml

@@ -5,6 +5,7 @@ status:
   class: receiver
   stability:
     beta: [metrics]
+    development: [logs]
   distributions: [core, contrib]
   codeowners:
     active: [dmitryax, braydonk]

receiver/hostmetricsreceiver/receiver.go

@@ -0,0 +1,110 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package hostmetricsreceiver
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/receiver"
+	"go.uber.org/zap"
+
+	metadataPkg "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/experimentalmetricmetadata"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent"
+)
+
+const entityType = "host"
+
+type hostMetricsReceiver struct {
+	cfg *Config
+
+	scraperHelper component.Component
+	nextLogs      consumer.Logs
+	shutdownWG    sync.WaitGroup
+	cancel        context.CancelFunc
+
+	settings *receiver.Settings
+}
+
+func (hmr *hostMetricsReceiver) Start(ctx context.Context, host component.Host) error {
+	ctx, hmr.cancel = context.WithCancel(ctx)
+	if hmr.nextLogs != nil {
+		hmr.sendEntityEvent(ctx)
+		if hmr.cfg.MetadataCollectionInterval != 0 {
+			ticker := time.NewTicker(hmr.cfg.MetadataCollectionInterval)
+			go func() {
+				for {
+					select {
+					case <-ticker.C:
+						hmr.sendEntityEvent(ctx)
+					case <-ctx.Done():
+						ticker.Stop()
+						return
+					}
+				}
+			}()
+		}
+	}
+
+	if hmr.scraperHelper != nil {
+		return hmr.scraperHelper.Start(ctx, host)
+	}
+	return nil
+}
+
+func (hmr *hostMetricsReceiver) Shutdown(ctx context.Context) error {
+	if hmr.cancel != nil {
+		hmr.cancel()
+	}
+	if hmr.scraperHelper != nil {
+		return hmr.scraperHelper.Shutdown(ctx)
+	}
+	return nil
+}
+
+func newHostMetricsReceiver(cfg *Config, set *receiver.Settings) (*hostMetricsReceiver, error) {
+	r := &hostMetricsReceiver{
+		cfg:      cfg,
+		nextLogs: nil,
+		settings: set,
+	}
+
+	return r, nil
+}
+
+func (hmr *hostMetricsReceiver) sendEntityEvent(ctx context.Context) {
+	timestamp := pcommon.NewTimestampFromTime(time.Now())
+
+	out := metadataPkg.NewEntityEventsSlice()
+	entityEvent := out.AppendEmpty()
+	entityEvent.SetTimestamp(timestamp)
+	state := entityEvent.SetEntityState()
+	state.SetEntityType(entityType)
+
+	logs := out.ConvertAndMoveToLogs()
+
+	err := hmr.nextLogs.ConsumeLogs(ctx, logs)
+	if err != nil {
+		hmr.settings.Logger.Error("Error sending entity event to the consumer", zap.Error(err))
+	}
+
+	// Note: receiver contract says that we need to retry sending if the
+	// returned error is not Permanent. However, we are not doing it here.
+	// Instead, we rely on the fact the metadata is collected periodically
+	// and the entity events will be delivered on the next cycle. This is
+	// fine because we deliver cumulative entity state.
+	// This allows us to avoid stressing the Collector or its destination
+	// unnecessarily (typically non-Permanent errors happen in stressed conditions).
+}
+
+// This is the map of already created hostmetric receivers for particular configurations.
+// We maintain this map because the Factory is asked log and metric receivers separately
+// when it gets CreateLogsReceiver() and CreateMetricsReceiver() but they must not
+// create separate objects, they must use one receiver object per configuration.
+var receivers = sharedcomponent.NewSharedComponents()