[Chrony Receiver] Part 2 - Internal chrony client (#12102)

* [ Chrony Receiver ] Adding internal components

Adding in the client with rewrites to allow usage with open telemetry
collector

* Allowing sequence to be based of time

Author: MovieStoreGuy

receiver/chronyreceiver/go.mod

@@ -3,6 +3,7 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/chrony
 go 1.17
 
 require (
+	github.com/facebook/time v0.0.0-20220713225404-f7a0d7702d50
 	github.com/stretchr/testify v1.8.0
 	github.com/tilinna/clock v1.1.0
 	go.opentelemetry.io/collector v0.56.0
@@ -22,8 +23,8 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/sirupsen/logrus v1.8.1 // indirect
 	go.opencensus.io v0.23.0 // indirect
 	go.opentelemetry.io/otel v1.8.0 // indirect
 	go.opentelemetry.io/otel/metric v0.31.0 // indirect

receiver/chronyreceiver/go.sum

@@ -0,0 +1,104 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed 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 chrony // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/chronyreceiver/internal/chrony"
+
+import (
+	"context"
+	"encoding/binary"
+	"errors"
+	"net"
+	"time"
+
+	"github.com/facebook/time/ntp/chrony"
+	"github.com/tilinna/clock"
+	"go.uber.org/multierr"
+)
+
+var (
+	errBadRequest = errors.New("bad request")
+)
+
+type Client interface {
+	// GetTrackingData will connection the configured chronyd endpoint
+	// and will read that instance tracking information relatively to the configured
+	// upstream NTP server(s).
+	GetTrackingData(ctx context.Context) (*Tracking, error)
+}
+
+type clientOption func(c *client)
+
+// client is a partial rewrite of the client provided by
+// github.com/facebook/time/ntp/chrony
+//
+// The reason for the partial rewrite is that the original
+// client uses logrus' global instance within the main code path.
+type client struct {
+	proto, addr string
+	timeout     time.Duration
+	dialer      func(network, addr string) (net.Conn, error)
+}
+
+// New creates a client ready to use with chronyd
+func New(addr string, timeout time.Duration, opts ...clientOption) (Client, error) {
+	if timeout < 1 {
+		return nil, errors.New("timeout must be positive")
+	}
+
+	network, endpoint, err := SplitNetworkEndpoint(addr)
+	if err != nil {
+		return nil, err
+	}
+
+	c := &client{
+		proto:   network,
+		addr:    endpoint,
+		timeout: timeout,
+		dialer:  net.Dial,
+	}
+	for _, opt := range opts {
+		opt(c)
+	}
+
+	return c, nil
+}
+
+func (c *client) GetTrackingData(ctx context.Context) (*Tracking, error) {
+	sock, err := c.dialer(c.proto, c.addr)
+	if err != nil {
+		return nil, err
+	}
+	clk := clock.FromContext(ctx)
+
+	if err = sock.SetDeadline(clk.Now().Add(c.timeout)); err != nil {
+		return nil, err
+	}
+
+	packet := chrony.NewTrackingPacket()
+	packet.SetSequence(uint32(clk.Now().UnixNano()))
+
+	if err := binary.Write(sock, binary.BigEndian, packet); err != nil {
+		return nil, multierr.Combine(err, sock.Close())
+	}
+	data := make([]uint8, 1024)
+	if _, err := sock.Read(data); err != nil {
+		return nil, multierr.Combine(err, sock.Close())
+	}
+
+	if err := sock.Close(); err != nil {
+		return nil, err
+	}
+
+	return newTrackingData(data)
+}

receiver/chronyreceiver/internal/chrony/client.go

@@ -0,0 +1,197 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed 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 chrony
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+	"net"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newMockConn(tb testing.TB, handler func(net.Conn) error) net.Conn {
+	client, server := net.Pipe()
+	tb.Cleanup(func() {
+		assert.NoError(tb, server.Close(), "Must not error when closing server connection")
+	})
+	assert.NoError(tb, server.SetDeadline(time.Now().Add(time.Second)), "Must not error when assigning deadline")
+	go func() {
+		assert.NoError(tb, binary.Read(server, binary.BigEndian, &requestTrackingContent{}), "Must not error when reading binary data")
+		assert.NoError(tb, handler(server), "Must not error when processing request")
+	}()
+	return client
+}
+
+func TestNew(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		scenario string
+		addr     string
+		toError  bool
+	}{
+		{scenario: "valid host", addr: "udp://localhost:323", toError: false},
+		{scenario: "missing port", addr: "udp://localhost", toError: true},
+		{scenario: "missing protocol", addr: "localhost:323", toError: true},
+		{scenario: "existing socket to potentially connect to", addr: fmt.Sprint("unix://", t.TempDir()), toError: false},
+		{scenario: "invalid path", addr: "unix:///no/socket", toError: true},
+		{scenario: "invalid protocol", addr: "http:/localhost:323", toError: true},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.scenario, func(t *testing.T) {
+			t.Parallel()
+			cl, err := New(tc.addr, time.Second)
+			if tc.toError {
+				assert.Error(t, err, "Must error")
+				assert.Nil(t, cl, "Must have a nil client")
+			} else {
+				assert.NoError(t, err, "Must not error")
+				assert.NotNil(t, cl, "Must have a client")
+			}
+		})
+	}
+}
+
+func TestGettingTrackingData(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		scenario string
+		handler  func(conn net.Conn) error
+		timeout  time.Duration
+		data     *Tracking
+		err      error
+	}{
+		{
+			scenario: "Successful read binary from socket",
+			timeout:  10 * time.Second,
+			handler: func(conn net.Conn) error {
+				type response struct {
+					ReplyHead
+					replyTrackingContent
+				}
+				resp := &response{
+					ReplyHead: ReplyHead{
+						Version: 6,
+						Status:  successfulRequest,
+						Reply:   replyTrackingCode,
+					},
+					replyTrackingContent: replyTrackingContent{
+						RefID: 100,
+						IPAddr: ipAddr{
+							IP:     [16]uint8{127, 0, 0, 1},
+							Family: ipAddrInet4,
+						},
+						Stratum:    10,
+						LeapStatus: 0,
+						RefTime: timeSpec{
+							100, 10, 0,
+						},
+						CurrentCorrection:  binaryFloat(1300),
+						LastOffset:         binaryFloat(10000),
+						RMSOffset:          binaryFloat(12000),
+						FreqPPM:            binaryFloat(3300),
+						ResidFreqPPM:       binaryFloat(123456),
+						SkewPPM:            binaryFloat(9943),
+						RootDelay:          binaryFloat(-1220),
+						RootDispersion:     binaryFloat(-1100000),
+						LastUpdateInterval: binaryFloat(120),
+					},
+				}
+
+				return binary.Write(conn, binary.BigEndian, resp)
+			},
+			data: &Tracking{
+				RefID:              100,
+				IPAddr:             net.IP([]byte{127, 0, 0, 1}),
+				Stratum:            10,
+				LeapStatus:         0,
+				RefTime:            (&timeSpec{100, 10, 0}).Time(),
+				CurrentCorrection:  binaryFloat(1300).Float(),
+				LastOffset:         binaryFloat(10000).Float(),
+				RMSOffset:          binaryFloat(12000).Float(),
+				FreqPPM:            binaryFloat(3300).Float(),
+				ResidFreqPPM:       binaryFloat(123456).Float(),
+				SkewPPM:            binaryFloat(9943).Float(),
+				RootDelay:          binaryFloat(-1220).Float(),
+				RootDispersion:     binaryFloat(-1100000).Float(),
+				LastUpdateInterval: binaryFloat(120).Float(),
+			},
+			err: nil,
+		},
+		{
+			scenario: "Timeout waiting for response",
+			timeout:  10 * time.Millisecond,
+			handler: func(conn net.Conn) error {
+				time.Sleep(100 * time.Millisecond)
+				return nil
+			},
+			err: os.ErrDeadlineExceeded,
+		},
+		{
+			scenario: "invalid status returned",
+			timeout:  5 * time.Second,
+			handler: func(conn net.Conn) error {
+				resp := &ReplyHead{
+					Version: 6,
+					Status:  1,
+					Reply:   replyTrackingCode,
+				}
+				return binary.Write(conn, binary.BigEndian, resp)
+			},
+			err: errBadRequest,
+		},
+		{
+			scenario: "invalid status command",
+			timeout:  5 * time.Second,
+			handler: func(conn net.Conn) error {
+				resp := &ReplyHead{
+					Version: 6,
+					Status:  successfulRequest,
+					Reply:   0,
+				}
+				return binary.Write(conn, binary.BigEndian, resp)
+			},
+			err: errBadRequest,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.scenario, func(t *testing.T) {
+			t.Parallel()
+			conn := newMockConn(t, tc.handler)
+
+			client, err := New(fmt.Sprintf("unix://%s", t.TempDir()), tc.timeout, func(c *client) {
+				c.dialer = func(_, _ string) (net.Conn, error) {
+					return conn, nil
+				}
+			})
+			require.NoError(t, err, "Must not error when creating client")
+
+			data, err := client.GetTrackingData(context.Background())
+			assert.EqualValues(t, tc.data, data, "Must match the expected data")
+			assert.ErrorIs(t, err, tc.err, "Must match the expected error")
+		})
+	}
+}