[receiver/ntp] add initial implementation (#35850)

#### Description
Adds initial implementation of ntpreceiver.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
#34375

receiver/ntpreceiver/config.go

@@ -4,6 +4,11 @@
 package ntpreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/ntpreceiver"
 
 import (
+	"errors"
+	"fmt"
+	"net"
+	"time"
+
 	"go.opentelemetry.io/collector/receiver/scraperhelper"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/ntpreceiver/internal/metadata"
@@ -13,5 +18,19 @@ import (
 type Config struct {
 	scraperhelper.ControllerConfig `mapstructure:",squash"`
 	metadata.MetricsBuilderConfig  `mapstructure:",squash"`
+	Version                        int    `mapstructure:"version"`
 	Endpoint                       string `mapstructure:"endpoint"`
 }
+
+func (c *Config) Validate() error {
+	var errs []error
+	_, _, err := net.SplitHostPort(c.Endpoint)
+	if err != nil {
+		errs = append(errs, err)
+	}
+	// respect terms of service https://www.pool.ntp.org/tos.html
+	if c.ControllerConfig.CollectionInterval < 30*time.Minute {
+		errs = append(errs, fmt.Errorf("collection interval %v is less than minimum 30m", c.ControllerConfig.CollectionInterval))
+	}
+	return errors.Join(errs...)
+}

receiver/ntpreceiver/config_test.go

@@ -0,0 +1,65 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ntpreceiver
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/receiver/scraperhelper"
+)
+
+func TestValidate(t *testing.T) {
+	for _, tt := range []struct {
+		name          string
+		c             *Config
+		errorExpected string
+	}{
+		{
+			name: "no host",
+			c: &Config{
+				Version:          4,
+				Endpoint:         "",
+				ControllerConfig: scraperhelper.ControllerConfig{CollectionInterval: 45 * time.Minute},
+			},
+			errorExpected: "missing port in address",
+		},
+		{
+			name: "no port",
+			c: &Config{
+				Version:          4,
+				Endpoint:         "pool.ntp.org",
+				ControllerConfig: scraperhelper.ControllerConfig{CollectionInterval: 45 * time.Minute},
+			},
+			errorExpected: "address pool.ntp.org: missing port in address",
+		},
+		{
+			name: "valid",
+			c: &Config{
+				Version:          4,
+				Endpoint:         "pool.ntp.org:123",
+				ControllerConfig: scraperhelper.ControllerConfig{CollectionInterval: 45 * time.Minute},
+			},
+		},
+		{
+			name: "interval too small",
+			c: &Config{
+				Version:          4,
+				Endpoint:         "pool.ntp.org:123",
+				ControllerConfig: scraperhelper.ControllerConfig{CollectionInterval: 29 * time.Minute},
+			},
+			errorExpected: "collection interval 29m0s is less than minimum 30m",
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.c.Validate()
+			if tt.errorExpected == "" {
+				require.NoError(t, err)
+			} else {
+				require.EqualError(t, err, tt.errorExpected)
+			}
+		})
+	}
+}

receiver/ntpreceiver/documentation.md

@@ -14,11 +14,11 @@ metrics:
 
 ### ntp.offset
 
-Time difference between local and NTP server clocks in seconds.
+Time difference between local and NTP server clocks
 
 | Unit | Metric Type | Value Type |
 | ---- | ----------- | ---------- |
-| s | Gauge | Int |
+| ns | Gauge | Int |
 
 ## Resource Attributes
 

receiver/ntpreceiver/factory.go

@@ -5,6 +5,7 @@ package ntpreceiver // import "github.com/open-telemetry/opentelemetry-collector
 
 import (
 	"context"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
@@ -23,12 +24,29 @@ func NewFactory() receiver.Factory {
 }
 
 func createDefaultConfig() component.Config {
+	scraperConfig := scraperhelper.NewDefaultControllerConfig()
+	scraperConfig.CollectionInterval = 30 * time.Minute
 	return &Config{
-		ControllerConfig:     scraperhelper.NewDefaultControllerConfig(),
+		ControllerConfig:     scraperConfig,
 		MetricsBuilderConfig: metadata.DefaultMetricsBuilderConfig(),
+		Version:              4,
+		Endpoint:             "pool.ntp.org:123",
 	}
 }
 
-func createMetricsReceiver(_ context.Context, _ receiver.Settings, _ component.Config, _ consumer.Metrics) (receiver.Metrics, error) {
-	return nil, nil
+func createMetricsReceiver(_ context.Context, settings receiver.Settings, cfg component.Config, consumer consumer.Metrics) (receiver.Metrics, error) {
+	rCfg := cfg.(*Config)
+	mp := newScraper(rCfg, settings)
+	s, err := scraperhelper.NewScraper(metadata.Type, mp.scrape)
+	if err != nil {
+		return nil, err
+	}
+	opt := scraperhelper.AddScraper(s)
+
+	return scraperhelper.NewScraperControllerReceiver(
+		&rCfg.ControllerConfig,
+		settings,
+		consumer,
+		opt,
+	)
 }

receiver/ntpreceiver/factory_test.go

@@ -0,0 +1,18 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ntpreceiver
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCreateDefaultConfig(t *testing.T) {
+	c := createDefaultConfig().(*Config)
+	require.Equal(t, 4, c.Version)
+	require.Equal(t, "pool.ntp.org:123", c.Endpoint)
+	require.Equal(t, 30*time.Minute, c.CollectionInterval)
+}

receiver/ntpreceiver/go.mod

@@ -3,11 +3,13 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/ntprec
 go 1.22.0
 
 require (
+	github.com/beevik/ntp v1.4.3
 	github.com/google/go-cmp v0.6.0
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/component v0.111.1-0.20241008154146-ea48c09c31ae
 	go.opentelemetry.io/collector/confmap v1.17.1-0.20241008154146-ea48c09c31ae
 	go.opentelemetry.io/collector/consumer v0.111.1-0.20241008154146-ea48c09c31ae
+	go.opentelemetry.io/collector/consumer/consumertest v0.111.1-0.20241008154146-ea48c09c31ae
 	go.opentelemetry.io/collector/filter v0.111.1-0.20241008154146-ea48c09c31ae
 	go.opentelemetry.io/collector/pdata v1.17.1-0.20241008154146-ea48c09c31ae
 	go.opentelemetry.io/collector/receiver v0.111.1-0.20241008154146-ea48c09c31ae
@@ -34,7 +36,6 @@ require (
 	github.com/rogpeppe/go-internal v1.12.0 // indirect
 	go.opentelemetry.io/collector/config/configtelemetry v0.111.1-0.20241008154146-ea48c09c31ae // indirect
 	go.opentelemetry.io/collector/consumer/consumerprofiles v0.111.1-0.20241008154146-ea48c09c31ae // indirect
-	go.opentelemetry.io/collector/consumer/consumertest v0.111.1-0.20241008154146-ea48c09c31ae // indirect
 	go.opentelemetry.io/collector/internal/globalsignal v0.111.1-0.20241008154146-ea48c09c31ae // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.111.1-0.20241008154146-ea48c09c31ae // indirect
 	go.opentelemetry.io/collector/pipeline v0.111.1-0.20241008154146-ea48c09c31ae // indirect

receiver/ntpreceiver/metadata.yaml

@@ -16,12 +16,10 @@ resource_attributes:
 
 metrics:
   ntp.offset:
-    description: Time difference between local and NTP server clocks in seconds.
-    unit: "s"
+    description: Time difference between local and NTP server clocks
+    unit: "ns"
     gauge:
       value_type: int
     enabled: true
 
-tests:
-  skip_lifecycle: true
-  skip_shutdown: true
+tests:

receiver/ntpreceiver/receiver.go

@@ -0,0 +1,46 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package ntpreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/ntpreceiver"
+
+import (
+	"context"
+	"time"
+
+	"github.com/beevik/ntp"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/receiver"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/ntpreceiver/internal/metadata"
+)
+
+type scraper struct {
+	logger   *zap.Logger
+	mb       *metadata.MetricsBuilder
+	version  int
+	timeout  time.Duration
+	endpoint string
+}
+
+func (s *scraper) scrape(context.Context) (pmetric.Metrics, error) {
+	options := ntp.QueryOptions{Version: s.version, Timeout: s.timeout}
+	response, err := ntp.QueryWithOptions(s.endpoint, options)
+	if err != nil {
+		return pmetric.Metrics{}, err
+	}
+	s.mb.RecordNtpOffsetDataPoint(pcommon.NewTimestampFromTime(time.Now()), response.ClockOffset.Nanoseconds())
+	s.mb.NewResourceBuilder().SetNtpHost(s.endpoint)
+	return s.mb.Emit(), nil
+}
+
+func newScraper(cfg *Config, settings receiver.Settings) *scraper {
+	return &scraper{
+		logger:   settings.TelemetrySettings.Logger,
+		mb:       metadata.NewMetricsBuilder(cfg.MetricsBuilderConfig, settings),
+		version:  cfg.Version,
+		timeout:  cfg.ControllerConfig.Timeout,
+		endpoint: cfg.Endpoint,
+	}
+}