Repository: clouditor/clouditor
Pull request #586 (2022-03-16 14:18:32)

Separate grpc calls
Decouple handleEvidence and sending to other components

- Add channel for evidences and assessment results
- Move grpc recv/ send evidences from/to Evidence Store in separate go routines
- Move grpc recv/send assessment results from/to Orchestrator in separate go routines
- Update tests

codecov[bot]: # [Codecov](https://codecov.io/gh/clouditor/clouditor/pull/586?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor) Report
> Merging [#586](https://codecov.io/gh/clouditor/clouditor/pull/586?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor) (c0bb608) into [main](https://codecov.io/gh/clouditor/clouditor/commit/c0351a22a4949831aeaae03d8a9ceea5f41c5b2a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor) (c0351a2) will **increase** coverage by `0.27%`.
> The diff coverage is `88.70%`.

```diff
@@            Coverage Diff             @@
##             main     #586      +/-   ##
==========================================
+ Coverage   82.16%   82.43%   +0.27%     
==========================================
  Files          58       58              
  Lines        5131     5165      +34     
==========================================
+ Hits         4216     4258      +42     
+ Misses        702      696       -6     
+ Partials      213      211       -2     
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `82.43% <88.70%> (+0.27%)` | :arrow_up: |

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/586?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor) | Coverage Δ | |
|---|---|---|
| [service/assessment/assessment.go](https://codecov.io/gh/clouditor/clouditor/pull/586/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor#diff-c2VydmljZS9hc3Nlc3NtZW50L2Fzc2Vzc21lbnQuZ28=) | `88.92% <88.70%> (+3.80%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/clouditor/clouditor/pull/586?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/clouditor/clouditor/pull/586?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor). Last update [c0351a2...c0bb608](https://codecov.io/gh/clouditor/clouditor/pull/586?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clouditor).

garuppel: Where do we want to start the go routines for the communication with the Evidence Store and Orchestrator (initXXX methods)?  Should we start that in the assessment.NewService()? 

Patch:
diff --git a/service/assessment/assessment.go b/service/assessment/assessment.go
index e2a562bc..660c109d 100644
--- a/service/assessment/assessment.go
+++ b/service/assessment/assessment.go
@@ -51,7 +51,9 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
-var log *logrus.Entry
+var (
+	log *logrus.Entry
+)
 
 func init() {
 	log = logrus.WithField("component", "assessment")
@@ -76,11 +78,15 @@ type Service struct {
 	// evidenceStoreStream sends evidences to the Evidence Store
 	evidenceStoreStream  evidence.EvidenceStore_StoreEvidencesClient
 	evidenceStoreAddress string
+	// evidenceStoreChannel stores evidences for the Evidence Store
+	evidenceStoreChannel chan *evidence.Evidence
 
-	// orchestratorStream sends ARs to the Orchestrator
+	// orchestratorStream sends assessment results to the Orchestrator
 	orchestratorStream  orchestrator.Orchestrator_StoreAssessmentResultsClient
 	orchestratorClient  orchestrator.OrchestratorClient
 	orchestratorAddress string
+	// orchestratorChannel stores assessment results for the Orchestrator
+	orchestratorChannel chan *assessment.AssessmentResult
 
 	// resultHooks is a list of hook functions that can be used if one wants to be
 	// informed about each assessment result
@@ -138,7 +144,9 @@ func NewService(opts ...ServiceOption) *Service {
 	s := &Service{
 		results:              make(map[string]*assessment.AssessmentResult),
 		evidenceStoreAddress: DefaultEvidenceStoreAddress,
+		evidenceStoreChannel: make(chan *evidence.Evidence, 1000),
 		orchestratorAddress:  DefaultOrchestratorAddress,
+		orchestratorChannel:  make(chan *assessment.AssessmentResult, 1000),
 		cachedConfigurations: make(map[string]cachedConfiguration),
 	}
 
@@ -147,6 +155,18 @@ func NewService(opts ...ServiceOption) *Service {
 		o(s)
 	}
 
+	// Initialise Evidence Store stream
+	err := s.initEvidenceStoreStream()
+	if err != nil {
+		log.Errorf("Error while initializing evidence store stream: %v", err)
+	}
+
+	// Initialise Orchestrator stream
+	err = s.initOrchestratorStream()
+	if err != nil {
+		log.Errorf("Error while initializing orchestrator stream: %v", err)
+	}
+
 	return s
 }
 
@@ -162,6 +182,8 @@ func (s *Service) Authorizer() api.Authorizer {
 
 // AssessEvidence is a method implementation of the assessment interface: It assesses a single evidence
 func (s *Service) AssessEvidence(_ context.Context, req *assessment.AssessEvidenceRequest) (res *assessment.AssessEvidenceResponse, err error) {
+
+	// Validate evidence
 	resourceId, err := req.Evidence.Validate()
 	if err != nil {
 		newError := fmt.Errorf("invalid evidence: %w", err)
@@ -178,7 +200,6 @@ func (s *Service) AssessEvidence(_ context.Context, req *assessment.AssessEviden
 
 	// Assess evidence
 	err = s.handleEvidence(req.Evidence, resourceId)
-
 	if err != nil {
 		res = &assessment.AssessEvidenceResponse{
 			Status:        assessment.AssessEvidenceResponse_FAILED,
@@ -191,6 +212,7 @@ func (s *Service) AssessEvidence(_ context.Context, req *assessment.AssessEviden
 		return res, status.Errorf(codes.Internal, "%v", newError)
 	}
 
+	// Create response
 	res = &assessment.AssessEvidenceResponse{
 		Status: assessment.AssessEvidenceResponse_ASSESSED,
 	}
@@ -206,8 +228,7 @@ func (s *Service) AssessEvidences(stream assessment.Assessment_AssessEvidencesSe
 	)
 
 	for {
-
-		// TODO(all): Check context?
+		// Receive requests from client
 		req, err = stream.Recv()
 
 		// If no more input of the stream is available, return
@@ -259,12 +280,8 @@ func (s *Service) handleEvidence(evidence *evidence.Evidence, resourceId string)
 		return newError
 	}
 
-	// The evidence is sent to the evidence store since it has been successfully evaluated
-	// We could also just log, but an AR could be sent without an accompanying evidence in the Evidence Store
-	err = s.sendToEvidenceStore(evidence)
-	if err != nil {
-		return fmt.Errorf("could not send evidence to the evidence store: %v", err)
-	}
+	// Store evidence in evidenceStoreChannel
+	s.evidenceStoreChannel <- evidence
 
 	for i, data := range evaluations {
 		metricId := data.MetricId
@@ -300,12 +317,10 @@ func (s *Service) handleEvidence(evidence *evidence.Evidence, resourceId string)
 		// Inform hooks about new assessment result
 		go s.informHooks(result, nil)
 
-		// Send assessment result to the Orchestrator
-		err = s.sendToOrchestrator(result)
-		if err != nil {
-			return fmt.Errorf("could not send assessment result to orchestrator: %v", err)
-		}
+		// Store assessment result in orchestratorChannel
+		s.orchestratorChannel <- result
 	}
+
 	return nil
 }
 
@@ -375,42 +390,57 @@ func (s *Service) initEvidenceStoreStream(additionalOpts ...grpc.DialOption) err
 		return fmt.Errorf("could not set up stream for storing evidences: %w", err)
 	}
 
+	log.Infof("Connected to Evidence Store")
+
 	// Receive responses from Evidence Store
 	// Currently we do not process the responses
 	go func() {
+		i := 1
 		for {
 			_, err := s.evidenceStoreStream.Recv()
+
 			if errors.Is(err, io.EOF) {
+				log.Debugf("no more responses from evidence store stream: EOF")
 				break
 			}
 
 			if err != nil {
-				log.Errorf("error receiving response from evidence store stream: %+v", err)
+				newError := fmt.Errorf("error receiving response from evidence store stream: %w", err)
+				log.Error(newError)
 				break
 			}
+
+			if i%100 == 0 {
+				log.Tracef("evidenceStoreStream recv responses currently @ %v", i)
+			}
+
+			i++
 		}
 	}()
 
-	log.Infof("Connected to Evidence Store")
+	// Send evidences from evidenceStoreChannel to the Evidence Store
+	go func() {
+		i := 1
+		for e := range s.evidenceStoreChannel {
+			err := s.evidenceStoreStream.Send(&evidence.StoreEvidenceRequest{Evidence: e})
+			if errors.Is(err, io.EOF) {
+				log.Debugf("EOF")
+				break
+			}
+			if err != nil {
+				log.Errorf("Error when sending evidence to Evidence Store:- %v", err)
+				break
+			}
 
-	return nil
-}
+			log.Debugf("Evidence (%v) sent to Evidence Store", e.Id)
 
-// sendToEvidenceStore sends evidence e to the Evidence Store
-func (s *Service) sendToEvidenceStore(e *evidence.Evidence) error {
-	if s.evidenceStoreStream == nil {
-		err := s.initEvidenceStoreStream()
-		if err != nil {
-			return fmt.Errorf("could not initialize stream to Evidence Store: %v", err)
+			if i%100 == 0 {
+				log.Debugf("evidenceStoreStream send evidences currently @ %v", i)
+			}
+			i++
 		}
-	}
-
-	log.Debugf("Sending evidence (%v) to Evidence Store", e.Id)
+	}()
 
-	err := s.evidenceStoreStream.Send(&evidence.StoreEvidenceRequest{Evidence: e})
-	if err != nil {
-		return err
-	}
 	return nil
 }
 
@@ -432,12 +462,18 @@ func (s *Service) initOrchestratorStream(additionalOpts ...grpc.DialOption) erro
 		return fmt.Errorf("could not set up stream for storing assessment results: %w", err)
 	}
 
+	log.Infof("Connected to Orchestrator")
+
 	// Receive responses from Orchestrator
 	// Currently we do not process the responses
 	go func() {
+		i := 1
+
 		for {
 			_, err := s.orchestratorStream.Recv()
+
 			if errors.Is(err, io.EOF) {
+				log.Debugf("no more responses from orchestrator stream: EOF")
 				break
 			}
 
@@ -445,33 +481,43 @@ func (s *Service) initOrchestratorStream(additionalOpts ...grpc.DialOption) erro
 				log.Errorf("error receiving response from orchestrator stream: %+v", err)
 				break
 			}
+
+			if i%100 == 0 {
+				log.Debugf("orchestratorStream recv responses currently @ %v", i)
+			}
+
+			i++
 		}
 	}()
 
-	log.Infof("Connected to Orchestrator")
+	// Send assessment results in orchestratorChannel to the Orchestrator
+	go func() {
+		i := 1
+		for result := range s.orchestratorChannel {
+			log.Debugf("Sending assessment result (%v) to Orchestrator", result.Id)
 
-	return nil
-}
+			req := &orchestrator.StoreAssessmentResultRequest{
+				Result: result,
+			}
 
-// sendToOrchestrator sends the assessment result to the Orchestrator
-func (s *Service) sendToOrchestrator(result *assessment.AssessmentResult) error {
-	if s.orchestratorStream == nil || s.orchestratorClient == nil {
-		err := s.initOrchestratorStream()
-		if err != nil {
-			return fmt.Errorf("could not initialize stream to Orchestrator: %v", err)
-		}
-	}
+			err := s.orchestratorStream.Send(req)
+			if errors.Is(err, io.EOF) {
+				log.Debugf("EOF")
+				break
+			}
+			if err != nil {
+				log.Errorf("Error when sending assessment result to Orchestrator: %v", err)
+				break
+			}
 
-	log.Debugf("Sending assessment result (%v) to Orchestrator", result.Id)
+			log.Debugf("Assessment result (%v) sent to Orchestrator", result.Id)
 
-	req := &orchestrator.StoreAssessmentResultRequest{
-		Result: result,
-	}
-	err := s.orchestratorStream.Send(req)
-	if err != nil {
-		log.Errorf("Error when sending assessment result to Orchestrator: %v", err)
-		return err
-	}
+			if i%100 == 0 {
+				log.Debugf("orchestratorStream send assessment results currently @ %v", i)
+			}
+			i++
+		}
+	}()
 
 	return nil
 }
diff --git a/service/discovery/discovery.go b/service/discovery/discovery.go
index 5366ba61..73377879 100644
--- a/service/discovery/discovery.go
+++ b/service/discovery/discovery.go
@@ -176,6 +176,7 @@ func (s *Service) initAssessmentStream(additionalOpts ...grpc.DialOption) error
 		for {
 			_, err := s.assessmentStream.Recv()
 			if errors.Is(err, io.EOF) {
+				log.Debugf("no more requests in assessment stream available")
 				break
 			}
 


Test patch:
diff --git a/service/assessment/assessment_test.go b/service/assessment/assessment_test.go
index ba7e9004..0e7df1c1 100644
--- a/service/assessment/assessment_test.go
+++ b/service/assessment/assessment_test.go
@@ -85,6 +85,8 @@ func TestNewService(t *testing.T) {
 				evidenceStoreAddress: "localhost:9090",
 				orchestratorAddress:  "localhost:9090",
 				cachedConfigurations: make(map[string]cachedConfiguration),
+				evidenceStoreChannel: nil,
+				orchestratorChannel:  nil,
 			},
 		},
 		{
@@ -105,7 +107,20 @@ func TestNewService(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := NewService(tt.args.opts...); !reflect.DeepEqual(got, tt.want) {
+			s := NewService(tt.args.opts...)
+
+			// Check channels have been created
+			assert.NotNil(t, s.evidenceStoreChannel)
+			assert.NotNil(t, s.orchestratorChannel)
+			assert.NotNil(t, s.orchestratorClient)
+
+			// Ignore pointers to channel in subsequent DeepEqual check
+			s.evidenceStoreChannel = nil
+			s.orchestratorChannel = nil
+			// Ignore pointer to orchestrator client in subsequent DeepEqual check
+			s.orchestratorClient = nil
+
+			if got := s; !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("NewService() = %v, want %v", got, tt.want)
 			}
 		})
@@ -193,24 +208,6 @@ func TestAssessEvidence(t *testing.T) {
 			},
 			wantErr: false,
 		},
-		{
-			name: "No RPC connections",
-			args: args{
-				in0: context.TODO(),
-				evidence: &evidence.Evidence{
-					Id:        "11111111-1111-1111-1111-111111111111",
-					ToolId:    "mock",
-					Timestamp: timestamppb.Now(),
-					Resource:  toStruct(voc.VirtualMachine{Compute: &voc.Compute{Resource: &voc.Resource{ID: "my-resource-id", Type: []string{"VirtualMachine"}}}}, t),
-				},
-			},
-			hasRPCConnection: false,
-			wantResp: &assessment.AssessEvidenceResponse{
-				Status:        assessment.AssessEvidenceResponse_FAILED,
-				StatusMessage: "could not evaluate evidence: could not fetch metric configuration: could not initialize connection to orchestrator: could not set up stream for storing assessment results: rpc error: code = Unavailable desc = connection error: desc = \"transport: Error while dialing dial tcp: missing address\"",
-			},
-			wantErr: true,
-		},
 	}
 
 	for _, tt := range tests {
@@ -372,6 +369,8 @@ func TestAssessEvidences(t *testing.T) {
 				results:                       tt.fields.results,
 				cachedConfigurations:          make(map[string]cachedConfiguration),
 				UnimplementedAssessmentServer: tt.fields.UnimplementedAssessmentServer,
+				evidenceStoreChannel:          make(chan *evidence.Evidence, 1000),
+				orchestratorChannel:           make(chan *assessment.AssessmentResult, 1000),
 			}
 
 			if tt.fields.hasRPCConnection {
@@ -764,7 +763,6 @@ func TestHandleEvidence(t *testing.T) {
 	type fields struct {
 		hasEvidenceStoreStream bool
 		hasOrchestratorStream  bool
-		//results                       map[string]*assessment.AssessmentResult
 	}
 	type args struct {
 		evidence   *evidence.Evidence
@@ -818,51 +816,6 @@ func TestHandleEvidence(t *testing.T) {
 				return true
 			},
 		},
-		{
-			name: "no connection to Evidence Store",
-			fields: fields{
-				hasOrchestratorStream:  true,
-				hasEvidenceStoreStream: false,
-			},
-			args: args{
-				evidence: &evidence.Evidence{
-					Id:        "11111111-1111-1111-1111-111111111111",
-					ToolId:    "mock",
-					Timestamp: timestamppb.Now(),
-					Resource:  toStruct(voc.VirtualMachine{Compute: &voc.Compute{Resource: &voc.Resource{ID: "my-resource-id", Type: []string{"VirtualMachine"}}}}, t),
-				},
-				resourceId: "my-resource-id",
-			},
-			wantErr: func(t assert.TestingT, err error, i ...interface{}) bool {
-				assert.Error(t, err)
-				// Check if error message contains "empty" (list of types)
-				assert.Contains(t, err.Error(), "Evidence Store")
-				assert.Contains(t, err.Error(), "Unavailable desc")
-				return true
-			},
-		},
-		{
-			name: "no connection to Orchestrator",
-			fields: fields{
-				hasOrchestratorStream:  false,
-				hasEvidenceStoreStream: true,
-			},
-			args: args{
-				evidence: &evidence.Evidence{
-					Id:        "11111111-1111-1111-1111-111111111111",
-					ToolId:    "mock",
-					Timestamp: timestamppb.Now(),
-					Resource:  toStruct(voc.VirtualMachine{Compute: &voc.Compute{Resource: &voc.Resource{ID: "my-resource-id", Type: []string{"VirtualMachine"}}}}, t),
-				},
-				resourceId: "my-resource-id",
-			},
-			wantErr: func(t assert.TestingT, err error, i ...interface{}) bool {
-				assert.Error(t, err)
-				// Check if error message contains "empty" (list of types)
-				assert.Contains(t, err.Error(), "Unavailable desc")
-				return true
-			},
-		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
diff --git a/service/assessment/benchmark_test.go b/service/assessment/benchmark_test.go
index 142e3aaa..c068cdc1 100644
--- a/service/assessment/benchmark_test.go
+++ b/service/assessment/benchmark_test.go
@@ -1,24 +1,23 @@
 package assessment
 
 import (
-	"context"
-	"fmt"
-	"net"
-	"net/http"
-	"sync"
-	"sync/atomic"
-	"testing"
-
 	"clouditor.io/clouditor/api/assessment"
 	"clouditor.io/clouditor/api/evidence"
 	"clouditor.io/clouditor/api/orchestrator"
 	service_evidence "clouditor.io/clouditor/service/evidence"
 	service_orchestrator "clouditor.io/clouditor/service/orchestrator"
 	"clouditor.io/clouditor/voc"
+	"context"
+	"fmt"
 	"github.com/google/uuid"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/timestamppb"
+	"net"
+	"net/http"
+	"sync"
+	"sync/atomic"
+	"testing"
 )
 
 func createEvidences(n int, m int, b *testing.B) int {
@@ -58,16 +57,6 @@ func createEvidences(n int, m int, b *testing.B) int {
 	addr := fmt.Sprintf("localhost:%d", sock.Addr().(*net.TCPAddr).Port)
 
 	svc := NewService(WithOrchestratorAddress(addr), WithEvidenceStoreAddress(addr))
-	// init our streams, otherwise we will have even more concurrency mess
-	err = svc.initEvidenceStoreStream()
-	if err != nil {
-		b.Fatalf("Error while initializing evidence store stream: %v", err)
-	}
-
-	err = svc.initOrchestratorStream()
-	if err != nil {
-		b.Fatalf("Error while initializing orchestrator stream: %v", err)
-	}
 
 	svc.RegisterAssessmentResultHook(func(result *assessment.AssessmentResult, err error) {
 		wg.Done()
@@ -75,7 +64,7 @@ func createEvidences(n int, m int, b *testing.B) int {
 		log.Debugf("Current count: %v", current)
 	})
 
-	// Create m parallel exeuctions of our evidence creation
+	// Create m parallel executions of our evidence creation
 	for j := 0; j < m; j++ {
 		go func() {
 			// Create evidences for n resources (1 per resource)