Repository: clouditor/clouditor
Pull request #977 (2023-02-21 21:36:42)

Re-starting old stream channels instead of re-creating them
This PR adds the behaviour of the `StreamsOf` when a channel is broken (e.g. through EOF). Previously, the stream channel was removed and completly created anew. However, this also meant that certain messages that were present at the time of breakage were lost. This PR changes this by putting the message back into the send-queue and recycling the send-queue in the new stream channel.


codecov[bot]: # [Codecov](https://codecov.io/gh/clouditor/clouditor/pull/977?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor) Report
> Merging [#977](https://codecov.io/gh/clouditor/clouditor/pull/977?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor) (5002720) into [main](https://codecov.io/gh/clouditor/clouditor/commit/f3ac6d86cdbe80d3d822bbce7f904614bfce758d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor) (f3ac6d8) will **decrease** coverage by `0.05%`.
> The diff coverage is `78.57%`.

```diff
@@            Coverage Diff             @@
##             main     #977      +/-   ##
==========================================
- Coverage   81.91%   81.86%   -0.05%     
==========================================
  Files          89       89              
  Lines        7524     7554      +30     
==========================================
+ Hits         6163     6184      +21     
- Misses       1044     1050       +6     
- Partials      317      320       +3     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `81.86% <78.57%> (-0.05%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/clouditor/clouditor/pull/977?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor) | Coverage Δ | |
|---|---|---|
| [api/streams.go](https://codecov.io/gh/clouditor/clouditor/pull/977?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor#diff-YXBpL3N0cmVhbXMuZ28=) | `88.43% <78.57%> (-4.73%)` | :arrow_down: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor)


Patch:
diff --git a/api/streams.go b/api/streams.go
index 448b54ae..7135bc47 100644
--- a/api/streams.go
+++ b/api/streams.go
@@ -60,6 +60,9 @@ type StreamChannelOf[StreamType grpc.ClientStream, MsgType proto.Message] struct
 
 	// component name
 	component string
+
+	// dead specifies that this channel lost connection and needs to be re-started
+	dead bool
 }
 
 // InitFuncOf describes a function with type parameters that creates any kind of stream towards a gRPC server specified
@@ -127,6 +130,12 @@ func (s *StreamsOf[StreamType, MsgType]) GetStream(target string, component stri
 		if err != nil {
 			return nil, fmt.Errorf("could not add stream for %s with target '%s': %w", component, target, err)
 		}
+	} else if c.dead {
+		// We could have a dead stream that we need to restart. in this case, we can recycle a few things, e.g. the channel
+		c, err = s.restartStream(c, init, opts...)
+		if err != nil {
+			return nil, fmt.Errorf("could not restart stream for %s with target '%s': %w", c.component, c.target, err)
+		}
 	}
 
 	return c, nil
@@ -169,46 +178,67 @@ func (s *StreamsOf[StreamType, MsgType]) addStream(target string, component stri
 	return c, nil
 }
 
-// removeStream deletes the channel from the stream map.
-func (s *StreamsOf[StreamType, MsgType]) removeStream(target string) {
-	s.log.Debugf("Removing stream channel for target %s", target)
+// restartStream restarts a stream to the given component and starts a goroutine for sending messages from the channel to the given component
+func (s *StreamsOf[StreamType, MsgType]) restartStream(c *StreamChannelOf[StreamType, MsgType], init InitFuncOf[StreamType], opts ...grpc.DialOption) (*StreamChannelOf[StreamType, MsgType], error) {
+	var err error
 
-	s.mutex.Lock()
-	delete(s.channels, target)
-	s.mutex.Unlock()
+	// We need an init func
+	if init == nil {
+		return nil, ErrMissingInitFunc
+	}
+
+	// Initialize the stream using our init function
+	c.stream, err = init(c.target, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("could not init stream: %w", err)
+	}
+
+	// Revive the stream
+	c.dead = false
+
+	s.log.Infof("Re-Established stream to %s (%s)", c.component, c.target)
+
+	// Start go routine for receiving messages from the stream (especially relevant for bi-directional streams).
+	go c.recvLoop(s)
+
+	// Start go routine for sending messages from the channel to the stream
+	go c.sendLoop(s)
+
+	return c, nil
 }
 
 // sendLoop continuously fetches new messages from the channel inside c and sends them to the appropriate stream.
 func (c *StreamChannelOf[StreamType, MsgType]) sendLoop(s *StreamsOf[StreamType, MsgType]) {
 	var err error
 
-	// Fetch new messages from channel (this will block)
+	// Fetch new (or queued old) messages from the channel. This will block.
 	for m := range c.channel {
+		// We want to log some additional information about this stream and its
+		// payload. The logging functions are safe to call with a nil request,
+		// so we can avoid checking, if this succeeds
+		preq, _ := any(m).(PayloadRequest)
+
 		// Try to send the message in our stream
 		err = c.stream.SendMsg(m)
-		if errors.Is(err, io.EOF) {
-			s.log.Infof("Stream to %s (%s) closed with EOF", c.component, c.target)
-
-			// Remove the stream from our map and end this goroutine
-			s.removeStream(c.target)
-			return
-		}
-
-		// Some other error than EOF occurred
 		if err != nil {
-			s.log.Errorf("Error when sending message to %s (%s): %v", c.component, c.target, err)
-
-			// Close the stream gracefully. We can ignore any error resulting from the close here
-			_ = c.stream.CloseSend()
-
-			// Remove the stream from our map and end this goroutine
-			s.removeStream(c.target)
-			return
-		}
-
-		// We want to log some additional information about this stream and its payload
-		preq, ok := any(m).(PayloadRequest)
-		if !ok {
+			if errors.Is(err, io.EOF) {
+				s.log.Infof("Stream to %s (%s) closed with EOF", c.component, c.target)
+			} else {
+				// Some other error than EOF occurred
+				s.log.Errorf("Error when sending message to %s (%s): %v", c.component, c.target, err)
+
+				// Close the stream gracefully. We can ignore any error resulting from the close here
+				_ = c.stream.CloseSend()
+			}
+
+			// Declare the stream as dead
+			c.dead = true
+
+			// Put the message back on the channel, so that it does not get lost
+			go func() {
+				logging.LogRequest(s.log, logrus.DebugLevel, logging.Store, preq, fmt.Sprintf("back into queue for %s (%s)", c.component, c.target))
+				c.channel <- m
+			}()
 			return
 		}
 
diff --git a/internal/logging/logging.go b/internal/logging/logging.go
index 6eed4971..02609429 100644
--- a/internal/logging/logging.go
+++ b/internal/logging/logging.go
@@ -26,6 +26,7 @@
 package logging
 
 import (
+	"bytes"
 	"fmt"
 	"strings"
 
@@ -91,7 +92,7 @@ func (r RequestType) String() string {
 //	"*orchestrator.TargetOfEvaluation created with ID 'ToE1234' for Cloud Service '00000000-0000-0000-0000-000000000000' and Catalog 'EUCS'."
 func LogRequest(log *logrus.Entry, level logrus.Level, reqType RequestType, req api.PayloadRequest, params ...string) {
 	var (
-		message string
+		buffer bytes.Buffer
 	)
 
 	// Check if inputs are available
@@ -114,21 +115,25 @@ func LogRequest(log *logrus.Entry, level logrus.Level, reqType RequestType, req
 	// Check, if our payload has an ID field
 	idreq, ok := payload.(interface{ GetId() string })
 	if ok && idreq.GetId() != "" {
-		message = fmt.Sprintf("%s with ID '%s' %s", name, idreq.GetId(), reqType.String())
+		buffer.WriteString(fmt.Sprintf("%s with ID '%s' %s", name, idreq.GetId(), reqType.String()))
 	} else {
-		message = fmt.Sprintf("%s %s", name, reqType.String())
+		buffer.WriteString(fmt.Sprintf("%s %s", name, reqType.String()))
 	}
 
 	// Check, if it is a cloud service request. In this case we can append the
 	// information about the target cloud service. However, we only want to do
 	// that, if the payload type is not a cloud service itself.
 	csreq, ok := req.(api.CloudServiceRequest)
+	if name != "CloudService" && ok {
+		buffer.WriteString(fmt.Sprintf(" for Cloud Service '%s'", csreq.GetCloudServiceId()))
+	}
+
 	// If params is not empty, the elements are joined and added to the message
-	if name != "CloudService" && ok && len(params) > 0 {
-		message = fmt.Sprintf("%s for Cloud Service '%s' %s", message, csreq.GetCloudServiceId(), strings.Join(params, " "))
-	} else if name != "CloudService" && ok {
-		message = fmt.Sprintf("%s for Cloud Service '%s'", message, csreq.GetCloudServiceId())
+	if len(params) > 0 {
+		buffer.WriteString(" " + strings.Join(params, " "))
 	}
 
-	log.Logf(level, "%s.", message)
+	buffer.WriteString(".")
+
+	log.Log(level, buffer.String())
 }


Test patch:
diff --git a/api/streams_test.go b/api/streams_test.go
index fc2ff0fd..91b4ba73 100644
--- a/api/streams_test.go
+++ b/api/streams_test.go
@@ -29,6 +29,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	sync "sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -38,6 +39,8 @@ import (
 	"google.golang.org/protobuf/reflect/protoreflect"
 
 	"clouditor.io/clouditor/api/assessment"
+	"clouditor.io/clouditor/api/evidence"
+	"clouditor.io/clouditor/internal/testdata"
 )
 
 var ErrSomeError = errors.New("some error")
@@ -47,16 +50,17 @@ func TestStreamsOf_GetStream(t *testing.T) {
 		target         string
 		component      string
 		additionalOpts []grpc.DialOption
-		init           InitFuncOf[*mockClientStream]
+		init           InitFuncOf[*recordedClientStream]
 	}
 	type fields struct {
-		channels map[string]*StreamChannelOf[*mockClientStream, proto.Message]
+		channels map[string]*StreamChannelOf[*recordedClientStream, proto.Message]
 	}
 	tests := []struct {
-		name    string
-		args    args
-		fields  fields
-		wantErr assert.ErrorAssertionFunc
+		name     string
+		args     args
+		fields   fields
+		wantErr  assert.ErrorAssertionFunc
+		wantRcvd int
 	}{
 		{
 			name:   "missing init function",
@@ -75,7 +79,7 @@ func TestStreamsOf_GetStream(t *testing.T) {
 			args: args{
 				target:    "localhost",
 				component: "mycomponent",
-				init: func(target string, additionalOpts ...grpc.DialOption) (m *mockClientStream, err error) {
+				init: func(target string, additionalOpts ...grpc.DialOption) (m *recordedClientStream, err error) {
 					return nil, ErrSomeError
 				},
 			},
@@ -88,28 +92,63 @@ func TestStreamsOf_GetStream(t *testing.T) {
 			args: args{
 				target:    "mock:1234",
 				component: "mock component",
-				init: func(target string, additionalOpts ...grpc.DialOption) (m *mockClientStream, err error) {
-					return &mockClientStream{}, nil
+				init: func(target string, additionalOpts ...grpc.DialOption) (m *recordedClientStream, err error) {
+					return &recordedClientStream{}, nil
 				},
 			},
 			fields: fields{
-				channels: map[string]*StreamChannelOf[*mockClientStream, proto.Message]{},
+				channels: map[string]*StreamChannelOf[*recordedClientStream, proto.Message]{},
 			},
 		},
+		{
+			name: "restarting stream",
+			fields: fields{
+				channels: map[string]*StreamChannelOf[*recordedClientStream, proto.Message]{
+					"mock:1234": {
+						dead: true,
+						channel: func() chan proto.Message {
+							// put 2 left over messages into the channel
+							var c = make(chan proto.Message)
+							go func() {
+								c <- &assessment.AssessEvidenceRequest{Evidence: &evidence.Evidence{Id: testdata.MockEvidenceID}}
+							}()
+							go func() {
+								c <- &assessment.AssessEvidenceRequest{Evidence: &evidence.Evidence{Id: testdata.MockAnotherEvidenceID}}
+							}()
+							return c
+						}(),
+						target:    "mock:1234",
+						component: "mock",
+					},
+				},
+			},
+			args: args{
+				target:    "mock:1234",
+				component: "mock component",
+				init: func(target string, additionalOpts ...grpc.DialOption) (m *recordedClientStream, err error) {
+					return &recordedClientStream{}, nil
+				},
+			},
+			wantRcvd: 2,
+		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			s := &StreamsOf[*mockClientStream, proto.Message]{
+			s := &StreamsOf[*recordedClientStream, proto.Message]{
 				channels: tt.fields.channels,
 				log:      defaultLog(),
 			}
-			_, err := s.GetStream(tt.args.target, tt.args.component, tt.args.init, tt.args.additionalOpts...)
+			c, err := s.GetStream(tt.args.target, tt.args.component, tt.args.init, tt.args.additionalOpts...)
 
 			if tt.wantErr != nil {
 				tt.wantErr(t, err, tt.args)
 			} else {
 				assert.Nil(t, err)
+
+				// wait until our stream has received the wanted messages
+				c.stream.wg.Add(tt.wantRcvd)
+				c.stream.wg.Wait()
 			}
 		})
 	}
@@ -150,8 +189,8 @@ func Test_StreamChannelOf_sendLoop(t *testing.T) {
 			want: func(tt assert.TestingT, i1 interface{}, i2 ...interface{}) bool {
 				s := i1.(*StreamsOf[*mockClientStream, proto.Message])
 
-				// sendLoop should remove itself from the channels on error
-				return assert.Empty(t, s.channels)
+				// sendLoop should declare the channel dead
+				return assert.True(t, s.channels["test"].dead)
 			},
 		},
 		{
@@ -173,8 +212,8 @@ func Test_StreamChannelOf_sendLoop(t *testing.T) {
 			want: func(tt assert.TestingT, i1 interface{}, i2 ...interface{}) bool {
 				s := i1.(*StreamsOf[*mockClientStream, proto.Message])
 
-				// sendLoop should remove itself from the channels on error
-				return assert.Empty(t, s.channels)
+				// sendLoop should declare the channel dead
+				return assert.True(t, s.channels["test"].dead)
 			},
 		},
 	}
@@ -188,6 +227,9 @@ func Test_StreamChannelOf_sendLoop(t *testing.T) {
 				component: tt.fields.component,
 			}
 
+			// overwrite the stream channel to make sure we are dealing with the same stream object
+			tt.args.s.channels["test"] = c
+
 			// prepare something, otherwise the sendloop will block waiting for a message
 			go func() { c.Send(&mockMessage{}) }()
 
@@ -245,3 +287,15 @@ func (m mockClientStream) SendMsg(interface{}) error {
 func (mockClientStream) RecvMsg(interface{}) error {
 	return nil
 }
+
+type recordedClientStream struct {
+	mockClientStream
+	recvd []proto.Message
+	wg    sync.WaitGroup
+}
+
+func (r *recordedClientStream) SendMsg(msg interface{}) error {
+	r.recvd = append(r.recvd, msg.(proto.Message))
+	r.wg.Done()
+	return nil
+}
diff --git a/internal/logging/logging_test.go b/internal/logging/logging_test.go
index 000bf3ef..f29c9f2e 100644
--- a/internal/logging/logging_test.go
+++ b/internal/logging/logging_test.go
@@ -30,6 +30,8 @@ import (
 	"fmt"
 	"testing"
 
+	"clouditor.io/clouditor/api/assessment"
+	"clouditor.io/clouditor/api/evidence"
 	"clouditor.io/clouditor/api/orchestrator"
 	"clouditor.io/clouditor/internal/api"
 	"clouditor.io/clouditor/internal/testdata"
@@ -170,6 +172,16 @@ func TestLogRequest(t *testing.T) {
 			},
 			want: "level=debug msg=TargetOfEvaluation updated for Cloud Service '11111111-1111-1111-1111-111111111111' and Catalog 'Cat1234'.\n",
 		},
+		{
+			name: "Send Evidence to queue",
+			args: args{
+				level:   logrus.DebugLevel,
+				reqType: Store,
+				req:     &assessment.AssessEvidenceRequest{Evidence: &evidence.Evidence{Id: testdata.MockEvidenceID}},
+				params:  []string{fmt.Sprintf("back into queue for %s (%s)", "orchestrator", "localhost")},
+			},
+			want: "level=debug msg=Evidence with ID '11111111-1111-1111-1111-111111111111' stored back into queue for orchestrator (localhost).\n",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {