Feature: Add metrics and uPlot chart for viewing metrics

Author: karthiknatarajan

.local.env

@@ -1,4 +1,7 @@
 CLOUDNESS_TRACE=true
 CLOUDNESS_ENVIRONMENT=local
 CLOUDNESS_ACME_USE_STAGING=true
-# CLOUDNESS_PUBSUB_PROVIDER=redis
+# CLOUDNESS_PUBSUB_PROVIDER=redis
+
+# CLOUDNESS_DATABASE_DRIVER=postgres
+# CLOUDNESS_DATABASE_DATASOURCE=postgres://postgres:postgres@localhost:5432/cloudness?sslmode=disable

app/controller/application/controller.go

@@ -27,6 +27,7 @@ type Controller struct {
 	schemaSvc        *schema.Service
 	specSvc          *spec.Service
 	applicationStore store.ApplicationStore
+	metricsStore     store.MetricsStore
 	serverCtrl       *server.Controller
 	varCtrl          *variable.Controller
 	gitPublicCtrl    *gitpublic.Controller
@@ -42,6 +43,7 @@ func NewController(
 	schemaSvc *schema.Service,
 	specSvc *spec.Service,
 	applicationStore store.ApplicationStore,
+	metricsStore store.MetricsStore,
 	serverCtrl *server.Controller,
 	varCtrl *variable.Controller,
 	gitPublicCtrl *gitpublic.Controller,
@@ -56,6 +58,7 @@ func NewController(
 		schemaSvc:        schemaSvc,
 		specSvc:          specSvc,
 		applicationStore: applicationStore,
+		metricsStore:     metricsStore,
 		serverCtrl:       serverCtrl,
 		varCtrl:          varCtrl,
 		gitPublicCtrl:    gitPublicCtrl,

app/controller/application/metrics.go

@@ -0,0 +1,123 @@
+package application
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/cloudness-io/cloudness/types"
+	"github.com/cloudness-io/cloudness/types/enum"
+)
+
+func (c *Controller) GetMetrics(ctx context.Context, application *types.Application, span enum.MetricsSpan) (*types.AppMetricsViewModel, error) {
+	from, to, bucket := getTimeFromSpan(span)
+
+	metrics, err := c.metricsStore.ListByApplicationUID(ctx, application.UID, from, to, bucket)
+	if err != nil {
+		return nil, err
+	}
+
+	return convertToMetricsView(application, metrics), nil
+}
+
+func convertToMetricsView(application *types.Application, metrics []*types.AppMetricsAggregate) *types.AppMetricsViewModel {
+	cpuInstanceData := make(map[string]map[int64]float64)
+	memInstanceData := make(map[string]map[int64]float64)
+	allTimestamps := make(map[int64]bool)
+	instanceNameMap := make(map[string]string)
+	var bytesToMB float64 = 1024 * 1024
+
+	for _, db := range metrics {
+		if _, ok := instanceNameMap[db.InstanceName]; !ok {
+			instanceNameMap[db.InstanceName] = fmt.Sprintf("%s-%d", application.Name, len(instanceNameMap))
+		}
+		instanceName, _ := instanceNameMap[db.InstanceName]
+		if _, ok := cpuInstanceData[instanceName]; !ok {
+			cpuInstanceData[instanceName] = make(map[int64]float64)
+			memInstanceData[instanceName] = make(map[int64]float64)
+		}
+		cpuInstanceData[instanceName][db.BucketTimestamp] = db.CPU
+		memInstanceData[instanceName][db.BucketTimestamp] = db.Memory
+		allTimestamps[db.BucketTimestamp] = true
+	}
+
+	// Create sorted timestamp array
+	timestamps := make([]int64, 0, len(allTimestamps))
+	for t := range allTimestamps {
+		timestamps = append(timestamps, t)
+	}
+
+	// Sort timestamps
+	for i := 0; i < len(timestamps)-1; i++ {
+		for j := i + 1; j < len(timestamps); j++ {
+			if timestamps[i] > timestamps[j] {
+				timestamps[i], timestamps[j] = timestamps[j], timestamps[i]
+			}
+		}
+	}
+
+	// Build CPU series
+	cpuSeries := make([]*types.MetricsSeriesViewModel, 0, len(cpuInstanceData))
+	for instanceName, data := range cpuInstanceData {
+		seriesValue := make([]float64, len(timestamps))
+		for i, ts := range timestamps {
+			if val, ok := data[ts]; ok {
+				seriesValue[i] = val
+			} else {
+				seriesValue[i] = 0
+			}
+		}
+
+		cpuSeries = append(cpuSeries, &types.MetricsSeriesViewModel{
+			Label:      instanceName,
+			Timestamps: timestamps,
+			Values:     seriesValue,
+		})
+	}
+
+	// Memory series
+	memSeries := make([]*types.MetricsSeriesViewModel, 0, len(memInstanceData))
+	for instanceName, data := range memInstanceData {
+		seriesValue := make([]float64, len(timestamps))
+		for i, ts := range timestamps {
+			if val, ok := data[ts]; ok {
+				seriesValue[i] = val / float64(bytesToMB)
+			} else {
+				seriesValue[i] = 0
+			}
+		}
+
+		memSeries = append(memSeries, &types.MetricsSeriesViewModel{
+			Label:      instanceName,
+			Timestamps: timestamps,
+			Values:     seriesValue,
+		})
+	}
+
+	return &types.AppMetricsViewModel{
+		CPU:    cpuSeries,
+		Memory: memSeries,
+	}
+}
+
+func getTimeFromSpan(span enum.MetricsSpan) (from time.Time, to time.Time, bucket int64) {
+	to = time.Now().UTC()
+	switch span {
+	case enum.MetricsSpan1h:
+		from = to.Add(-1 * time.Hour)
+		bucket = 60 // 1 min
+	case enum.MetricsSpan6h:
+		from = to.Add(-6 * time.Hour)
+		bucket = 2 * 60 // 2 min
+	case enum.MetricsSpan1d:
+		from = to.Add(-24 * time.Hour)
+		bucket = 5 * 60 // 5 min
+	case enum.MetricsSpan7d:
+		from = to.Add(-7 * 24 * time.Hour)
+		bucket = 15 * 60 // 15 min
+		// case enum.MetricsSpan30d:
+		// 	from = to - 30*24*60*60
+		// 	bucket = 30 * 24 * 300
+	}
+	return
+}

app/controller/application/wire.go

@@ -28,6 +28,7 @@ func ProvideController(
 	schemaSvc *schema.Service,
 	specSvc *spec.Service,
 	applicationStore store.ApplicationStore,
+	metricsStore store.MetricsStore,
 	serverCtrl *server.Controller,
 	varCtrl *variable.Controller,
 	gitPublicCtrl *gitpublic.Controller,
@@ -42,6 +43,7 @@ func ProvideController(
 		schemaSvc,
 		specSvc,
 		applicationStore,
+		metricsStore,
 		serverCtrl,
 		varCtrl,
 		gitPublicCtrl,

app/controller/project/create.go

@@ -9,6 +9,7 @@ import (
 	"github.com/cloudness-io/cloudness/errors"
 	"github.com/cloudness-io/cloudness/helpers"
 	"github.com/cloudness-io/cloudness/types"
+	"github.com/cloudness-io/cloudness/types/enum"
 
 	"github.com/rs/zerolog/log"
 )
@@ -58,6 +59,12 @@ func (c *Controller) Create(ctx context.Context, session *auth.Session, tenant *
 			return err
 		}
 
+		if err := c.AddMember(ctx, session, tenant.ID, project.ID, &ProjectMembershipAddModel{
+			Email: session.Principal.Email,
+			Role:  enum.ProjectRoleOwner,
+		}); err != nil {
+			return err
+		}
 		_, err = c.envCtrl.Create(ctx, session, tenant, project, &environment.CreateEnvironmentInput{
 			Name: "Development",
 		})

app/pipeline/agent/agent.go

@@ -3,16 +3,15 @@ package agent
 import (
 	"context"
 	"fmt"
-	"os"
 	"runtime/debug"
-	"sync"
 	"time"
 
 	"github.com/cloudness-io/cloudness/app/pipeline/manager/client"
-	"github.com/cloudness-io/cloudness/app/pipeline/runner"
 	"github.com/cloudness-io/cloudness/app/pipeline/runner/engine"
 	"github.com/cloudness-io/cloudness/app/pipeline/runner/engine/kubernetes"
+	"github.com/cloudness-io/cloudness/app/services/manager"
 	"github.com/cloudness-io/cloudness/types"
+	"github.com/cloudness-io/cloudness/types/enum"
 
 	"github.com/rs/zerolog/log"
 )
@@ -24,13 +23,20 @@ var engines = []engine.Engine{
 type Agent struct {
 	serverUID     int64
 	client        client.RunnerClient
+	serverManager manager.ServerManager
 	stopChan      <-chan bool
 	configUpdated int64
 }
 
-func New(client client.RunnerClient) *Agent {
+func New(client client.RunnerClient, factory manager.ManagerFactory) *Agent {
+	serverManager, err := factory.GetServerManagerByType(enum.ServerTypeK8s)
+	if err != nil {
+		panic(err)
+	}
+
 	return &Agent{
-		client: client,
+		client:        client,
+		serverManager: serverManager,
 	}
 }
 
@@ -69,6 +75,8 @@ func (a *Agent) Start(ctx context.Context) error {
 					rCtxCancel() // cancels previous run
 				}
 				rCtx, rCtxCancel = context.WithCancel(ctx)
+
+				//Build Runner
 				if config.EnableRunner {
 					go func() {
 						if err := a.run(rCtx, *config); err != nil {
@@ -77,6 +85,16 @@ func (a *Agent) Start(ctx context.Context) error {
 						}
 					}()
 				}
+
+				//Metrics Scrapper
+				// if config.EnableMetricsScrapper{
+				go func() {
+					if err := a.runMetricsScrapper(rCtx, *config); err != nil {
+						log.Ctx(ctx).Error().Err(err).Msg("agent: error running metrics scrapper")
+						return
+					}
+				}()
+				// }
 			}
 		}
 	}()
@@ -107,128 +125,3 @@ func (a *Agent) checkConfigChange(ctx context.Context, configChan chan<- *types.
 		}
 	}
 }
-
-func (a *Agent) run(ctx context.Context, config types.RunnerConfig) error {
-	engine, err := engine.FindEngine(engines, config)
-	if err != nil {
-		log.Error().Err(err).Msg("agent: error finding engine")
-		return err
-	}
-
-	_, err = engine.Load(ctx, config)
-	if err != nil {
-		log.Ctx(ctx).Error().Err(err).Msgf("agent: error loading %s engine", engine.Type())
-		return err
-	}
-
-	hostname := config.Hostname
-	if len(hostname) == 0 {
-		hostname, _ = os.Hostname()
-	}
-
-	log.Info().Msgf("agent: starting agent with %s engine", engine.Type())
-
-	var wg sync.WaitGroup
-	jobChan := make(chan *types.Deployment)
-	defer close(jobChan)
-	// number of runners based on job
-	for i := range config.ParallelWorkers {
-		wg.Go(func() {
-			for {
-				select {
-				case <-ctx.Done():
-					return
-				case deployment := <-jobChan:
-					log := log.With().
-						Int64("agent.thread", i).
-						Int64("deployment.id", deployment.ID).
-						Int64("deployment.uid", deployment.UID).
-						Logger()
-					deployment.Machine = hostname
-					if err := a.client.Accept(ctx, deployment); err != nil {
-						log.Error().Err(err).Msg("agent: unable to accept deployment")
-						continue
-					}
-
-					log.Trace().Msg("agent: deployment accepted for processing")
-
-					runnerContext, err := a.client.Init(ctx, deployment.ID)
-					if err != nil {
-						log.Error().Err(err).Msg("agent: unable to initialize deployment")
-						continue
-					}
-					log.Trace().Msg("agent: deployment initialized for processing")
-					log = log.With().
-						Int64("application.uid", runnerContext.ApplicationUID).
-						Int64("deployment.uid", runnerContext.Deployment.UID).
-						Str("runner.id", runnerContext.RunnerName).
-						Logger()
-
-					runnerCtx, runnerCancel := context.WithCancel(log.WithContext(ctx))
-					runner := runner.NewRunner(a.client, config, engine)
-
-					go func() {
-						log.Debug().Msg("agent: watch for deployment cancellation")
-						done, _ := a.client.Watch(ctx, deployment.ID)
-						if done {
-							log.Debug().Msg("agent: received cancellation event")
-							runnerCancel()
-						} else {
-							log.Debug().Msg("agent: done listening for cancellation event")
-						}
-					}()
-
-					runner.Start(runnerCtx, runnerContext)
-				}
-			}
-		})
-	}
-
-	//listing incomplete
-	runner := runner.NewRunner(a.client, config, engine)
-	if incomplete, err := runner.ListPending(ctx); err != nil {
-		log.Ctx(ctx).Error().Err(err).Any("Incomplete", incomplete).Msg("agent: error listing pending jobs, skipping...")
-	} else {
-		if len(incomplete) == 0 {
-			log.Ctx(ctx).Info().Msg("agent: no pending jobs found")
-		} else {
-			log.Ctx(ctx).Info().Any("pending", len(incomplete)).Msg("agent: found pending jobs, TODO: yet to be done")
-		}
-	}
-
-	a.startPolling(ctx, config, jobChan)
-	wg.Wait()
-
-	return nil
-}
-
-func (a *Agent) startPolling(ctx context.Context, config types.RunnerConfig, jobChan chan<- *types.Deployment) {
-	ticker := time.NewTicker(time.Duration(config.PollingInterval) * time.Second)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-			deployment, err := a.client.Request(ctx)
-			if err == context.Canceled || err == context.DeadlineExceeded {
-				log.Trace().Err(err).Msg("agent: no deployment received")
-				continue
-			}
-			if err != nil {
-				log.Error().Err(err).Msg("agent: unable to request deployment")
-				//TODO: max retries?
-				continue
-			}
-
-			log.Trace().Msg("agent: deployment received")
-
-			if deployment == nil || deployment.ID == 0 {
-				continue
-			}
-
-			jobChan <- deployment
-		}
-	}
-}