chore(rollout-service): Adapt rollout service to get information from the GetAppDetails endpoint

pkg/api/v1/api.proto

@@ -378,13 +378,20 @@ service OverviewService {
   rpc GetOverview (GetOverviewRequest) returns (GetOverviewResponse) {}
   rpc StreamOverview (GetOverviewRequest) returns (stream GetOverviewResponse) {}
 
+  rpc StreamChangedApps (GetChangedAppsRequest) returns (stream GetChangedAppsResponse) {}
   rpc GetAppDetails (GetAppDetailsRequest) returns (GetAppDetailsResponse) {}
 }
 
 service EnvironmentService {
   rpc GetEnvironmentConfig(GetEnvironmentConfigRequest) returns (GetEnvironmentConfigResponse) {}
 }
 
+message GetChangedAppsRequest {}
+
+message GetChangedAppsResponse {
+  repeated string changed_apps = 1;
+}
+
 message GetAppDetailsRequest {
   string app_name = 1;
 }
@@ -396,6 +403,7 @@ message GetAppDetailsResponse {
   map<string, Locks> team_locks= 4; //EnvName -> []TeamLocks
 }
 
+
 //Wrapper over array of locks
 message Locks {
   repeated Lock locks = 2;
@@ -421,7 +429,6 @@ message Deployment {
   DeploymentMetaData deployment_meta_data = 7;
 }
 
-
 message GetOverviewRequest {
   // Retrieve the overview at a certain state of the repository. If it's empty, the latest commit will be used.
   string git_revision = 1;

services/cd-service/pkg/notify/notify.go

@@ -16,11 +16,14 @@ Copyright freiheit.com*/
 
 package notify
 
-import "sync"
+import (
+	"sync"
+)
 
 type Notify struct {
-	mx       sync.Mutex
-	listener map[chan struct{}]struct{}
+	mx                 sync.Mutex
+	listener           map[chan struct{}]struct{}
+	changeAppsListener map[chan []string][]string
 }
 
 type Unsubscribe = func()
@@ -53,3 +56,31 @@ func (n *Notify) Notify() {
 		}
 	}
 }
+
+func (n *Notify) SubscribeChangesApps() (<-chan []string, Unsubscribe) {
+	ch := make(chan []string, 1)
+	ch <- []string{}
+
+	n.mx.Lock()
+	defer n.mx.Unlock()
+	if n.changeAppsListener == nil {
+		n.changeAppsListener = map[chan []string][]string{}
+	}
+	n.changeAppsListener[ch] = []string{}
+	return ch, func() {
+		n.mx.Lock()
+		defer n.mx.Unlock()
+		delete(n.changeAppsListener, ch)
+	}
+}
+
+func (n *Notify) NotifyChangedApps(changedApps []string) {
+	n.mx.Lock()
+	defer n.mx.Unlock()
+	for ch := range n.changeAppsListener {
+		select {
+		case ch <- changedApps:
+		default:
+		}
+	}
+}

services/cd-service/pkg/repository/repository.go

@@ -601,7 +601,7 @@ func (r *repository) applyTransformerBatches(transformerBatches []transformerBat
 			}
 		}
 	}
-
+	
 	return transformerBatches, nil, changes
 }
 
@@ -740,7 +740,6 @@ func (r *repository) ProcessQueueOnce(ctx context.Context, e transformerBatch, c
 	}
 
 	transformerBatches, err, changes := r.applyTransformerBatches(transformerBatches, true)
-
 	if len(transformerBatches) == 0 {
 		return
 	}
@@ -794,6 +793,17 @@ func (r *repository) ProcessQueueOnce(ctx context.Context, e transformerBatch, c
 	ddSpan.Finish()
 
 	r.notify.Notify()
+	var changedAppNames []string
+	var seen = make(map[string]bool)
+	for _, app := range changes.ChangedApps {
+		if _, ok := seen[app.App]; !ok {
+			seen[app.App] = true
+			changedAppNames = append(changedAppNames, app.App)
+		}
+	}
+	if len(changedAppNames) != 0 {
+		r.notify.NotifyChangedApps(changedAppNames)
+	}
 }
 
 func UpdateDatadogMetricsDB(ctx context.Context, state *State, r Repository, changes *TransformerResult, now time.Time) error {
@@ -1122,6 +1132,18 @@ func (r *repository) Apply(ctx context.Context, transformers ...Transformer) err
 			}
 		}
 		r.notify.Notify()
+
+		var changedAppNames []string
+		var seen = make(map[string]bool)
+		for _, app := range changes.ChangedApps {
+			if _, ok := seen[app.App]; !ok {
+				seen[app.App] = true
+				changedAppNames = append(changedAppNames, app.App)
+			}
+		}
+		if len(changedAppNames) != 0 {
+			r.notify.NotifyChangedApps(changedAppNames)
+		}
 		return nil
 	} else {
 		eCh := r.applyDeferred(ctx, transformers...)
@@ -2210,7 +2232,6 @@ func (s *State) WriteCurrentEnvironmentLocks(ctx context.Context, transaction *s
 			return err
 		}
 		for lockId, lock := range ls {
-			fmt.Printf("LOCK: %s\n", lock.CreatedAt.String())
 			currentEnv := db.EnvironmentLock{
 				EslVersion: 0,
 				Env:        envName,

services/cd-service/pkg/repository/transformer.go

@@ -2792,6 +2792,17 @@ func (c *CreateEnvironment) Transform(
 		if err != nil {
 			return "", fmt.Errorf("Unable to write overview cache, error: %w", err)
 		}
+
+		//Should be empty on new environments
+		envApps, err := state.GetEnvironmentApplications(ctx, transaction, c.Environment)
+		if err != nil {
+			return "", fmt.Errorf("Unable to read environment, error: %w", err)
+
+		}
+		for _, app := range envApps {
+			t.AddAppEnv(app, c.Environment, "")
+		}
+
 	} else {
 		fs := state.Filesystem
 		envDir := fs.Join("environments", c.Environment)
@@ -3069,6 +3080,7 @@ func (c *DeployApplicationVersion) Transform(
 		if err != nil {
 			return "", fmt.Errorf("could not write oldest deployment for %v - %v", newDeployment, err)
 		}
+
 	} else {
 		//Check if there is a version of target app already deployed on target environment
 		if _, err := fs.Lstat(versionFile); err == nil {
@@ -3107,11 +3119,6 @@ func (c *DeployApplicationVersion) Transform(
 		if err := util.WriteFile(fs, manifestFilename, manifestContent, 0666); err != nil {
 			return "", err
 		}
-		teamOwner, err := state.GetApplicationTeamOwner(ctx, transaction, c.Application)
-		if err != nil {
-			return "", err
-		}
-		t.AddAppEnv(c.Application, c.Environment, teamOwner)
 
 		if err := util.WriteFile(fs, fs.Join(applicationDir, "deployed_by"), []byte(user.Name), 0666); err != nil {
 			return "", err
@@ -3124,7 +3131,11 @@ func (c *DeployApplicationVersion) Transform(
 			return "", err
 		}
 	}
-
+	teamOwner, err := state.GetApplicationTeamOwner(ctx, transaction, c.Application)
+	if err != nil {
+		return "", err
+	}
+	t.AddAppEnv(c.Application, c.Environment, teamOwner)
 	s := State{
 		Commit:               nil,
 		MinorRegexes:         state.MinorRegexes,

services/cd-service/pkg/service/overview.go

@@ -46,10 +46,11 @@ type OverviewServiceServer struct {
 	RepositoryConfig repository.RepositoryConfig
 	Shutdown         <-chan struct{}
 
-	notify   notify.Notify
-	Context  context.Context
-	init     sync.Once
-	response atomic.Value
+	notify                       notify.Notify
+	Context                      context.Context
+	init                         sync.Once
+	changedAppsStreamingInitFunc sync.Once
+	response                     atomic.Value
 
 	DBHandler *db.DBHandler
 }
@@ -349,6 +350,31 @@ func (o *OverviewServiceServer) StreamOverview(in *api.GetOverviewRequest,
 	}
 }
 
+func (o *OverviewServiceServer) StreamChangedApps(in *api.GetChangedAppsRequest,
+	stream api.OverviewService_StreamChangedAppsServer) error {
+	ch, unsubscribe := o.subscribeChangedApps()
+	defer unsubscribe()
+	done := stream.Context().Done()
+	for {
+		select {
+		case <-o.Shutdown:
+			return nil
+		case changedApps := <-ch:
+			ov := &api.GetChangedAppsResponse{
+				ChangedApps: changedApps,
+			}
+			logger.FromContext(stream.Context()).Sugar().Infof("Got changes apps: '%v'\n", changedApps)
+			if err := stream.Send(ov); err != nil {
+				logger.FromContext(stream.Context()).Error("error sending changed apps  response:", zap.Error(err), zap.String("changedApps", fmt.Sprintf("%+v", ov)))
+				return err
+			}
+
+		case <-done:
+			return nil
+		}
+	}
+}
+
 func (o *OverviewServiceServer) subscribe() (<-chan struct{}, notify.Unsubscribe) {
 	o.init.Do(func() {
 		ch, unsub := o.Repository.Notify().Subscribe()
@@ -372,6 +398,29 @@ func (o *OverviewServiceServer) subscribe() (<-chan struct{}, notify.Unsubscribe
 	return o.notify.Subscribe()
 }
 
+func (o *OverviewServiceServer) subscribeChangedApps() (<-chan []string, notify.Unsubscribe) {
+	o.changedAppsStreamingInitFunc.Do(func() {
+		ch, unsub := o.Repository.Notify().SubscribeChangesApps()
+		// Channels obtained from subscribe are by default triggered
+		//
+		// This means, we have to wait here until the first overview is loaded.
+		<-ch
+		o.notify.NotifyChangedApps([]string{})
+		go func() {
+			defer unsub()
+			for {
+				select {
+				case <-o.Shutdown:
+					return
+				case changedApps := <-ch:
+					o.notify.NotifyChangedApps(changedApps)
+				}
+			}
+		}()
+	})
+	return o.notify.SubscribeChangesApps()
+}
+
 func (o *OverviewServiceServer) update(s *repository.State) {
 	r, err := o.getOverviewDB(o.Context, s)
 	if err != nil {

services/frontend-service/pkg/cmd/server.go

@@ -704,6 +704,24 @@ func (p *GrpcProxy) StreamOverview(
 	}
 }
 
+func (p *GrpcProxy) StreamChangedApps(
+	in *api.GetChangedAppsRequest,
+	stream api.OverviewService_StreamChangedAppsServer) error {
+	if resp, err := p.OverviewClient.StreamChangedApps(stream.Context(), in); err != nil {
+		return err
+	} else {
+		for {
+			if item, err := resp.Recv(); err != nil {
+				return err
+			} else {
+				if err := stream.Send(item); err != nil {
+					return err
+				}
+			}
+		}
+	}
+}
+
 func (p *GrpcProxy) StreamStatus(in *api.StreamStatusRequest, stream api.RolloutService_StreamStatusServer) error {
 	if p.RolloutServiceClient == nil {
 		return status.Error(codes.Unimplemented, "rollout service not configured")