Implement the shipper gRPC server

NOTICE.txt

@@ -478,11 +478,11 @@ Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-l
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/elastic-agent-shipper-client
-Version: v0.3.0
+Version: v0.4.0
 Licence type (autodetected): Elastic
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-shipper-client@v0.3.0/LICENSE.txt:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-shipper-client@v0.4.0/LICENSE.txt:
 
 Elastic License 2.0
 
@@ -790,6 +790,36 @@ Contents of probable licence file $GOMODCACHE/github.com/elastic/go-ucfg@v0.8.6/
    limitations under the License.
 
 
+--------------------------------------------------------------------------------
+Dependency : github.com/gofrs/uuid
+Version: v4.2.0+incompatible
+Licence type (autodetected): MIT
+--------------------------------------------------------------------------------
+
+Contents of probable licence file $GOMODCACHE/github.com/gofrs/uuid@v4.2.0+incompatible/LICENSE:
+
+Copyright (C) 2013-2018 by Maxim Bublis <b@codemonkey.ru>
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+
 --------------------------------------------------------------------------------
 Dependency : github.com/magefile/mage
 Version: v1.13.0
@@ -25123,36 +25153,6 @@ OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 
---------------------------------------------------------------------------------
-Dependency : github.com/gofrs/uuid
-Version: v4.2.0+incompatible
-Licence type (autodetected): MIT
---------------------------------------------------------------------------------
-
-Contents of probable licence file $GOMODCACHE/github.com/gofrs/uuid@v4.2.0+incompatible/LICENSE:
-
-Copyright (C) 2013-2018 by Maxim Bublis <b@codemonkey.ru>
-
-Permission is hereby granted, free of charge, to any person obtaining
-a copy of this software and associated documentation files (the
-"Software"), to deal in the Software without restriction, including
-without limitation the rights to use, copy, modify, merge, publish,
-distribute, sublicense, and/or sell copies of the Software, and to
-permit persons to whom the Software is furnished to do so, subject to
-the following conditions:
-
-The above copyright notice and this permission notice shall be
-included in all copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
-EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
-MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
-NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
-LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
-OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
-WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
-
-
 --------------------------------------------------------------------------------
 Dependency : github.com/gogo/protobuf
 Version: v1.3.2

go.mod

@@ -13,8 +13,9 @@ require (
 require (
 	github.com/elastic/beats/v7 v7.0.0-alpha2.0.20220722175030-7cb39607b349
 	github.com/elastic/elastic-agent-client/v7 v7.0.0-20220607160924-1a71765a8bbe
-	github.com/elastic/elastic-agent-shipper-client v0.3.0
+	github.com/elastic/elastic-agent-shipper-client v0.4.0
 	github.com/elastic/go-ucfg v0.8.6
+	github.com/gofrs/uuid v4.2.0+incompatible
 	github.com/magefile/mage v1.13.0
 	github.com/stretchr/testify v1.7.1
 	go.elastic.co/go-licence-detector v0.5.0
@@ -32,7 +33,6 @@ require (
 	github.com/elastic/go-windows v1.0.1 // indirect
 	github.com/fatih/color v1.13.0 // indirect
 	github.com/gobuffalo/here v0.6.0 // indirect
-	github.com/gofrs/uuid v4.2.0+incompatible // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/licenseclassifier v0.0.0-20200402202327-879cb1424de0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect

go.sum

@@ -501,8 +501,8 @@ github.com/elastic/elastic-agent-libs v0.2.7/go.mod h1:chO3rtcLyGlKi9S0iGVZhYCzD
 github.com/elastic/elastic-agent-libs v0.2.9 h1:7jOCqNqEWG0kJb3fa8/SC6beSiys1TmAylH9+hWTnrM=
 github.com/elastic/elastic-agent-libs v0.2.9/go.mod h1:chO3rtcLyGlKi9S0iGVZhYCzDfdDsAQYBc+ui588AFE=
 github.com/elastic/elastic-agent-shipper-client v0.2.0/go.mod h1:OyI2W+Mv3JxlkEF3OeT7K0dbuxvwew8ke2Cf4HpLa9Q=
-github.com/elastic/elastic-agent-shipper-client v0.3.0 h1:Ec/r08WLB6tfR95lW9JUTCUYD+HcqQdK7MR7n+0ax4s=
-github.com/elastic/elastic-agent-shipper-client v0.3.0/go.mod h1:OyI2W+Mv3JxlkEF3OeT7K0dbuxvwew8ke2Cf4HpLa9Q=
+github.com/elastic/elastic-agent-shipper-client v0.4.0 h1:nsTJF9oo4RHLl+zxFUZqNHaE86C6Ba5aImfegcEf6Sk=
+github.com/elastic/elastic-agent-shipper-client v0.4.0/go.mod h1:OyI2W+Mv3JxlkEF3OeT7K0dbuxvwew8ke2Cf4HpLa9Q=
 github.com/elastic/elastic-agent-system-metrics v0.4.3/go.mod h1:tF/f9Off38nfzTZHIVQ++FkXrDm9keFhFpJ+3pQ00iI=
 github.com/elastic/elastic-transport-go/v8 v8.1.0/go.mod h1:87Tcz8IVNe6rVSLdBux1o/PEItLtyabHU3naC7IoqKI=
 github.com/elastic/go-concert v0.2.0/go.mod h1:HWjpO3IAEJUxOeaJOWXWEp7imKd27foxz9V5vegC/38=

queue/queue.go

@@ -77,8 +77,8 @@ func (queue *Queue) Get(eventCount int) (beatsqueue.Batch, error) {
 	return queue.eventQueue.Get(eventCount)
 }
 
-func (queue *Queue) Close() {
-	queue.eventQueue.Close()
+func (queue *Queue) Close() error {
+	return queue.eventQueue.Close()
 }
 
 func (queue *Queue) AcceptedIndex() EntryID {

server/controller_client_test.go

@@ -0,0 +1,112 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package server
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+
+	"github.com/elastic/elastic-agent-client/v7/pkg/client"
+	"github.com/elastic/elastic-agent-client/v7/pkg/client/mock"
+	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
+)
+
+func TestAgentControl(t *testing.T) {
+	unitOneID := mock.NewID()
+
+	token := mock.NewID()
+	var gotConfig, gotHealthy, gotStopped bool
+
+	var mut sync.Mutex
+
+	t.Logf("Creating mock server")
+	srv := mock.StubServerV2{
+		CheckinV2Impl: func(observed *proto.CheckinObserved) *proto.CheckinExpected {
+			mut.Lock()
+			defer mut.Unlock()
+			if observed.Token == token {
+				if len(observed.Units) > 0 {
+					t.Logf("Current unit state is: %v", observed.Units[0].State)
+				}
+
+				// initial checkin
+				if len(observed.Units) == 0 || observed.Units[0].State == proto.State_STARTING {
+					gotConfig = true
+					return &proto.CheckinExpected{
+						Units: []*proto.UnitExpected{
+							{
+								Id:             unitOneID,
+								Type:           proto.UnitType_OUTPUT,
+								ConfigStateIdx: 1,
+								Config:         `{"logging": {"level": "debug"}}`, // hack to make my life easier
+								State:          proto.State_HEALTHY,
+							},
+						},
+					}
+				} else if observed.Units[0].State == proto.State_HEALTHY {
+					gotHealthy = true
+					//shutdown
+					return &proto.CheckinExpected{
+						Units: []*proto.UnitExpected{
+							{
+								Id:             unitOneID,
+								Type:           proto.UnitType_OUTPUT,
+								ConfigStateIdx: 1,
+								Config:         "{}",
+								State:          proto.State_STOPPED,
+							},
+						},
+					}
+				} else if observed.Units[0].State == proto.State_STOPPED {
+					gotStopped = true
+					// remove the unit? I think?
+					return &proto.CheckinExpected{
+						Units: nil,
+					}
+				}
+
+			}
+
+			//gotInvalid = true
+			return nil
+		},
+		ActionImpl: func(response *proto.ActionResponse) error {
+
+			return nil
+		},
+		ActionsChan: make(chan *mock.PerformAction, 100),
+	} // end of srv declaration
+
+	require.NoError(t, srv.Start())
+	defer srv.Stop()
+
+	t.Logf("creating client")
+	// connect with client
+	validClient := client.NewV2(fmt.Sprintf(":%d", srv.Port), token, client.VersionInfo{
+		Name:    "program",
+		Version: "v1.0.0",
+		Meta: map[string]string{
+			"key": "value",
+		},
+	}, grpc.WithTransportCredentials(insecure.NewCredentials()))
+
+	t.Logf("starting shipper controller")
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+	err := runController(ctx, validClient)
+	assert.NoError(t, err)
+
+	assert.True(t, gotConfig, "config state")
+	assert.True(t, gotHealthy, "healthy state")
+	assert.True(t, gotStopped, "stopped state")
+}

server/run.go

@@ -104,11 +104,11 @@ func (c *clientHandler) Run(cfg config.ShipperConfig, unit *client.Unit) error {
 		opts = []grpc.ServerOption{grpc.Creds(creds)}
 	}
 	grpcServer := grpc.NewServer(opts...)
-	r := shipperServer{
-		logger: log,
-		queue:  queue,
+	shipperServer, err := NewShipperServer(queue)
+	if err != nil {
+		return fmt.Errorf("failed to initialise the server: %w", err)
 	}
-	pb.RegisterProducerServer(grpcServer, r)
+	pb.RegisterProducerServer(grpcServer, shipperServer)
 
 	shutdownFunc := func() {
 		grpcServer.GracefulStop()
@@ -118,6 +118,7 @@ func (c *clientHandler) Run(cfg config.ShipperConfig, unit *client.Unit) error {
 		// We call Wait to give it a chance to finish with events
 		// it has already read.
 		out.Wait()
+		shipperServer.Close()
 	}
 	handleShutdown(shutdownFunc, c.shutdownInit)
 	log.Debugf("gRPC server is listening on port %d", cfg.Port)