feat: implement paginated polling agent-side (#120)

* implement paginated polling agent-side

* gen mocks

* bump console client

* service context templating

Author: zreigz

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/orcaman/concurrent-map/v2 v2.0.1
 	github.com/osteele/liquid v1.3.2
 	github.com/pkg/errors v0.9.1
-	github.com/pluralsh/console-client-go v0.0.90
+	github.com/pluralsh/console-client-go v0.0.92
 	github.com/pluralsh/controller-reconcile-helper v0.0.4
 	github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34
 	github.com/pluralsh/polly v0.1.4

go.sum

@@ -579,8 +579,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZI=
 github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
-github.com/pluralsh/console-client-go v0.0.90 h1:dxtuW2C9ynhhs358ckx0XGH4zieaoMr9EGf+pJb4hO4=
-github.com/pluralsh/console-client-go v0.0.90/go.mod h1:eyCiLA44YbXiYyJh8303jk5JdPkt9McgCo5kBjk4lKo=
+github.com/pluralsh/console-client-go v0.0.92 h1:PMSF05Zp5gLejeEWXbwe17CfXNLJ55dGlFPAAVucfCM=
+github.com/pluralsh/console-client-go v0.0.92/go.mod h1:eyCiLA44YbXiYyJh8303jk5JdPkt9McgCo5kBjk4lKo=
 github.com/pluralsh/controller-reconcile-helper v0.0.4 h1:1o+7qYSyoeqKFjx+WgQTxDz4Q2VMpzprJIIKShxqG0E=
 github.com/pluralsh/controller-reconcile-helper v0.0.4/go.mod h1:AfY0gtteD6veBjmB6jiRx/aR4yevEf6K0M13/pGan/s=
 github.com/pluralsh/gophoenix v0.1.3-0.20231201014135-dff1b4309e34 h1:ab2PN+6if/Aq3/sJM0AVdy1SYuMAnq4g20VaKhTm/Bw=

pkg/client/console.go

@@ -67,12 +67,12 @@ type Client interface {
 	UpdateClusterRestore(id string, attrs console.RestoreAttributes) (*console.ClusterRestoreFragment, error)
 	SaveClusterBackup(attrs console.BackupAttributes) (*console.ClusterBackupFragment, error)
 	GetClusterBackup(clusterID, namespace, name string) (*console.ClusterBackupFragment, error)
-	GetServices() ([]*console.ServiceDeploymentBaseFragment, error)
+	GetServices(after *string, first *int64) (*console.PagedClusterServices, error)
 	GetService(id string) (*console.GetServiceDeploymentForAgent_ServiceDeployment, error)
 	UpdateComponents(id string, components []*console.ComponentAttributes, errs []*console.ServiceErrorAttributes) error
 	AddServiceErrors(id string, errs []*console.ServiceErrorAttributes) error
 	ParsePipelineGateCR(pgFragment *console.PipelineGateFragment) (*v1alpha1.PipelineGate, error)
 	GetClusterGate(id string) (*console.PipelineGateFragment, error)
-	GetClusterGates() ([]*console.PipelineGateFragment, error)
+	GetClusterGates(after *string, first *int64) (*console.PagedClusterGates, error)
 	UpdateGate(id string, attributes console.GateUpdateAttributes) error
 }

pkg/client/pipelines.go

@@ -17,13 +17,15 @@ func (c *client) UpdateGate(id string, attributes console.GateUpdateAttributes)
 	return err
 }
 
-func (c *client) GetClusterGates() ([]*console.PipelineGateFragment, error) {
-	resp, err := c.consoleClient.GetClusterGates(c.ctx)
+func (c *client) GetClusterGates(after *string, first *int64) (*console.PagedClusterGates, error) {
+	resp, err := c.consoleClient.PagedClusterGates(c.ctx, after, first, nil, nil)
 	if err != nil {
 		return nil, err
 	}
-
-	return resp.ClusterGates, nil
+	if resp.PagedClusterGates == nil {
+		return nil, fmt.Errorf("the response from PagedClusterGates is nil")
+	}
+	return resp, nil
 }
 
 func (c *client) GetClusterGate(id string) (*console.PipelineGateFragment, error) {

pkg/client/service.go

@@ -1,16 +1,20 @@
 package client
 
 import (
+	"fmt"
 	console "github.com/pluralsh/console-client-go"
 )
 
-func (c *client) GetServices() ([]*console.ServiceDeploymentBaseFragment, error) {
-	resp, err := c.consoleClient.ListClusterServices(c.ctx)
+func (c *client) GetServices(after *string, first *int64) (*console.PagedClusterServices, error) {
+
+	resp, err := c.consoleClient.PagedClusterServices(c.ctx, after, first, nil, nil)
 	if err != nil {
 		return nil, err
 	}
-
-	return resp.ClusterServices, nil
+	if resp.GetPagedClusterServices() == nil {
+		return nil, fmt.Errorf("the response from PagedClusterServices is nil")
+	}
+	return resp, nil
 }
 
 func (c *client) GetService(id string) (*console.GetServiceDeploymentForAgent_ServiceDeployment, error) {

pkg/controller/pipelinegates/reconciler.go

@@ -94,17 +94,27 @@ func (s *GateReconciler) ShutdownQueue() {
 
 func (s *GateReconciler) Poll(ctx context.Context) (done bool, err error) {
 	logger := log.FromContext(ctx)
-
 	logger.Info("fetching gates for cluster")
-	gates, err := s.ConsoleClient.GetClusterGates()
-	if err != nil {
-		logger.Error(err, "failed to fetch gates list")
-		return false, nil
-	}
 
-	for _, gate := range gates {
-		logger.Info("sending update for", "gate", gate.ID)
-		s.GateQueue.Add(gate.ID)
+	var after *string
+	var pageSize int64
+	pageSize = 100
+	hasNextPage := true
+
+	for hasNextPage {
+		resp, err := s.ConsoleClient.GetClusterGates(after, &pageSize)
+		if err != nil {
+			logger.Error(err, "failed to fetch gates list")
+			return false, nil
+		}
+
+		hasNextPage = resp.PagedClusterGates.PageInfo.HasNextPage
+		after = resp.PagedClusterGates.PageInfo.EndCursor
+
+		for _, gate := range resp.PagedClusterGates.Edges {
+			logger.Info("sending update for", "gate", gate.Node.ID)
+			s.GateQueue.Add(gate.Node.ID)
+		}
 	}
 
 	if err := s.pinger.Ping(); err != nil {

pkg/controller/service/reconciler.go

@@ -177,17 +177,25 @@ func (s *ServiceReconciler) ShutdownQueue() {
 
 func (s *ServiceReconciler) Poll(ctx context.Context) (done bool, err error) {
 	logger := log.FromContext(ctx)
-
 	logger.Info("fetching services for cluster")
-	svcs, err := s.ConsoleClient.GetServices()
-	if err != nil {
-		logger.Error(err, "failed to fetch service list from deployments service")
-		return false, nil
-	}
+	var after *string
+	var pageSize int64
+	pageSize = 100
+	hasNextPage := true
 
-	for _, svc := range svcs {
-		logger.Info("sending update for", "service", svc.ID)
-		s.SvcQueue.Add(svc.ID)
+	for hasNextPage {
+		resp, err := s.ConsoleClient.GetServices(after, &pageSize)
+		if err != nil {
+			logger.Error(err, "failed to fetch service list from deployments service")
+			return false, nil
+		}
+
+		hasNextPage = resp.PagedClusterServices.PageInfo.HasNextPage
+		after = resp.PagedClusterServices.PageInfo.EndCursor
+		for _, svc := range resp.PagedClusterServices.Edges {
+			logger.Info("sending update for", "service", svc.Node.ID)
+			s.SvcQueue.Add(svc.Node.ID)
+		}
 	}
 
 	if err := s.pinger.Ping(); err != nil {

pkg/manifests/template/raw.go

@@ -48,7 +48,7 @@ func renderLiquid(input []byte, svc *console.GetServiceDeploymentForAgent_Servic
 	bindings := map[string]interface{}{
 		"configuration": configMap(svc),
 		"cluster":       svc.Cluster,
-		"contexts":      map[string]map[string]interface{}{},
+		"contexts":      contexts(svc),
 	}
 	return liquidEngine.ParseAndRender(input, bindings)
 }

pkg/manifests/template/utils.go

@@ -12,3 +12,11 @@ func configMap(svc *console.GetServiceDeploymentForAgent_ServiceDeployment) map[
 
 	return res
 }
+
+func contexts(svc *console.GetServiceDeploymentForAgent_ServiceDeployment) map[string]map[string]interface{} {
+	res := map[string]map[string]interface{}{}
+	for _, context := range svc.Contexts {
+		res[context.Name] = context.Configuration
+	}
+	return res
+}