Stateful checks

api/context/context.go

@@ -7,13 +7,16 @@ import (
 	"strings"
 	"time"
 
+	"github.com/flanksource/canary-checker/api/external"
 	v1 "github.com/flanksource/canary-checker/api/v1"
+	"github.com/flanksource/canary-checker/pkg"
 	"github.com/flanksource/commons/logger"
 	ctemplate "github.com/flanksource/commons/template"
 	"github.com/flanksource/duty"
 	dutyCtx "github.com/flanksource/duty/context"
 	"github.com/flanksource/duty/models"
 	"github.com/flanksource/duty/types"
+	"github.com/flanksource/gomplate/v3"
 	"github.com/flanksource/kommons"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"gorm.io/gorm"
@@ -39,8 +42,9 @@ type Context struct {
 	Canary      v1.Canary
 	Environment map[string]interface{}
 	logger.Logger
-	db   *gorm.DB
-	pool *pgxpool.Pool
+	db    *gorm.DB
+	pool  *pgxpool.Pool
+	cache map[string]any
 }
 
 func (ctx *Context) DB() *gorm.DB {
@@ -80,6 +84,26 @@ func getDomain(username string) string {
 	return ""
 }
 
+func (ctx *Context) GetFunctionsFor(check external.Check) map[string]any {
+	env := make(map[string]any)
+
+	return env
+}
+
+func (ctx *Context) Template(check external.Check, template string) (string, error) {
+	env := ctx.Environment
+
+	for k, v := range ctx.GetFunctionsFor(check) {
+		env[k] = v
+	}
+
+	out, err := gomplate.RunExpression(env, gomplate.Template{Template: template})
+	if err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("%s", out), nil
+}
+
 func (ctx *Context) GetConnection(conn v1.Connection) (*models.Connection, error) {
 	var _conn *models.Connection
 	var err error
@@ -240,10 +264,44 @@ func (ctx *Context) Tracef(format string, args ...interface{}) {
 	}
 }
 
+func (ctx *Context) WithCheckResult(result *pkg.CheckResult) *Context {
+	ctx = ctx.WithCheck(result.Check)
+	ctx.Environment["duration"] = result.Duration
+	for k, v := range result.Data {
+		ctx.Environment[k] = v
+	}
+	return ctx
+}
+
+func (ctx *Context) WithCheck(check external.Check) *Context {
+	env := make(map[string]any)
+
+	checkID := ctx.Canary.GetCheckID(check.GetName())
+
+	env["canary"] = map[string]any{
+		"name":      ctx.Canary.GetName(),
+		"namespace": ctx.Canary.GetNamespace(),
+		"labels":    ctx.Canary.GetLabels(),
+		"id":        ctx.Canary.GetPersistedID(),
+	}
+
+	env["check"] = map[string]any{
+		"name":        check.GetName(),
+		"id":          checkID,
+		"description": check.GetDescription(),
+		"labels":      check.GetLabels(),
+		"endpoint":    check.GetEndpoint(),
+	}
+	return ctx.New(env)
+}
+
 func (ctx *Context) New(environment map[string]interface{}) *Context {
 	return &Context{
 		Context:     ctx.Context,
 		Kommons:     ctx.Kommons,
+		db:          ctx.db,
+		pool:        ctx.pool,
+		Kubernetes:  ctx.Kubernetes,
 		Namespace:   ctx.Namespace,
 		Canary:      ctx.Canary,
 		Environment: environment,

api/context/functions.go

@@ -0,0 +1,81 @@
+package context
+
+import (
+	"encoding/json"
+	"time"
+
+	"github.com/flanksource/commons/logger"
+	"github.com/flanksource/gomplate/v3"
+)
+
+func (ctx *Context) InjectFunctions(template *gomplate.Template) {
+	if check, ok := ctx.Environment["check"]; ok {
+		checkID := check.(map[string]any)["id"]
+		if template.Functions == nil {
+			template.Functions = make(map[string]func() any)
+		}
+		template.Functions["last_result"] = func() any {
+			if ctx.cache == nil {
+				ctx.cache = make(map[string]any)
+			}
+			if result, ok := ctx.cache["last_result"]; ok {
+				return result
+			}
+			var status map[string]any
+
+			if checkID == "" {
+				return status
+			}
+
+			if ctx.DB() == nil {
+				logger.Errorf("[last_result] db connection not initialized")
+				return status
+			}
+
+			type CheckStatus struct {
+				Status    bool      `json:"status"`
+				Invalid   bool      `json:"invalid,omitempty"`
+				Time      string    `json:"time" gorm:"primaryKey"`
+				Duration  int       `json:"duration"`
+				Message   string    `json:"message,omitempty"`
+				Error     string    `json:"error,omitempty"`
+				Details   string    `json:"details" gorm:"details"`
+				CreatedAt time.Time `json:"created_at,omitempty"`
+			}
+
+			var checkStatus CheckStatus
+			err := ctx.DB().
+				Table("check_statuses").
+				Select("status", "invalid", "time", "duration", "message", "error", "details", "created_at").
+				Where("check_id = ?", checkID).
+				Order("time DESC").Limit(1).Scan(&checkStatus).Error
+			if err != nil {
+				logger.Warnf("[last_result] failed => %s", err)
+				return status
+			}
+
+			status = map[string]any{
+				"status":    checkStatus.Status,
+				"invalid":   checkStatus.Invalid,
+				"createdAt": checkStatus.CreatedAt,
+				"duration":  checkStatus.Duration,
+				"message":   checkStatus.Message,
+				"error":     checkStatus.Error,
+				"results":   make(map[string]any),
+			}
+
+			if checkStatus.Details != "" {
+				var details = make(map[string]any)
+				if err := json.Unmarshal([]byte(checkStatus.Details), &details); err == nil {
+					status["results"] = details
+				} else {
+					if ctx.IsTrace() {
+						ctx.Warnf("[last_result] Failed to unmarshal results: %s", err.Error())
+					}
+				}
+			}
+			ctx.cache["last_result"] = status
+			return status
+		}
+	}
+}

checks/common.go

@@ -62,7 +62,11 @@ func unstructure(o any) (out interface{}, err error) {
 }
 
 func template(ctx *context.Context, template v1.Template) (string, error) {
-	return gomplate.RunTemplate(ctx.Environment, template.Gomplate())
+	tpl := template.Gomplate()
+
+	ctx.InjectFunctions(&tpl)
+
+	return gomplate.RunTemplate(ctx.Environment, tpl)
 }
 
 func transform(ctx *context.Context, in *pkg.CheckResult) ([]*pkg.CheckResult, error) {
@@ -76,7 +80,7 @@ func transform(ctx *context.Context, in *pkg.CheckResult) ([]*pkg.CheckResult, e
 		return []*pkg.CheckResult{in}, nil
 	}
 
-	out, err := template(ctx.New(in.Data), tpl)
+	out, err := template(ctx, tpl)
 	if err != nil {
 		return nil, err
 	}

checks/metrics.go

@@ -5,7 +5,6 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/flanksource/canary-checker/api/context"
 	"github.com/flanksource/canary-checker/api/external"
@@ -42,24 +41,6 @@ func getOrAddPrometheusMetric(name, metricType string, labelNames []string) (col
 	return collector, prometheus.Register(collector)
 }
 
-func getWithEnvironment(ctx *context.Context, r *pkg.CheckResult) *context.Context {
-	r.Data["canary"] = map[string]any{
-		"name":      r.Canary.GetName(),
-		"namespace": r.Canary.GetNamespace(),
-		"labels":    r.Canary.GetLabels(),
-		"id":        r.Canary.GetPersistedID(),
-	}
-	r.Data["check"] = map[string]any{
-		"name":        r.Check.GetName(),
-		"id":          r.Canary.GetCheckID(r.Check.GetName()),
-		"description": r.Check.GetDescription(),
-		"labels":      r.Check.GetLabels(),
-		"endpoint":    r.Check.GetEndpoint(),
-		"duration":    time.Millisecond * time.Duration(r.GetDuration()),
-	}
-	return ctx.New(r.Data)
-}
-
 func getLabels(ctx *context.Context, metric external.Metrics) (map[string]string, error) {
 	var labels = make(map[string]string)
 	for _, label := range metric.Labels {
@@ -107,8 +88,9 @@ func exportCheckMetrics(ctx *context.Context, results pkg.Results) {
 	}
 
 	for _, r := range results {
+		checkCtx := ctx.WithCheckResult(r)
 		for _, metric := range r.Metrics {
-			if err := exportMetric(ctx, metric); err != nil {
+			if err := exportMetric(checkCtx, metric); err != nil {
 				r.ErrorMessage(err)
 			}
 		}
@@ -117,11 +99,9 @@ func exportCheckMetrics(ctx *context.Context, results pkg.Results) {
 				continue
 			}
 
-			ctx = getWithEnvironment(ctx, r)
-
-			if metric, err := templateMetrics(ctx, spec); err != nil {
+			if metric, err := templateMetrics(checkCtx, spec); err != nil {
 				r.ErrorMessage(err)
-			} else if err := exportMetric(ctx, *metric); err != nil {
+			} else if err := exportMetric(checkCtx, *metric); err != nil {
 				r.ErrorMessage(err)
 			}
 		}

checks/runchecks.go

@@ -103,13 +103,14 @@ func RunChecks(ctx *context.Context) ([]*pkg.CheckResult, error) {
 
 func transformResults(ctx *context.Context, in []*pkg.CheckResult) (out []*pkg.CheckResult) {
 	for _, r := range in {
-		transformed, err := transform(ctx, r)
+		checkCtx := ctx.WithCheckResult(r)
+		transformed, err := transform(checkCtx, r)
 		if err != nil {
 			r.Failf("transformation failure: %v", err)
 			out = append(out, r)
 		} else {
 			for _, t := range transformed {
-				out = append(out, processTemplates(ctx, t))
+				out = append(out, processTemplates(checkCtx, t))
 			}
 		}
 	}
@@ -151,11 +152,10 @@ func processTemplates(ctx *context.Context, r *pkg.CheckResult) *pkg.CheckResult
 	if r.Duration == 0 && r.GetDuration() > 0 {
 		r.Duration = r.GetDuration()
 	}
-
 	switch v := r.Check.(type) {
 	case v1.DisplayTemplate:
 		if !v.GetDisplayTemplate().IsEmpty() {
-			message, err := template(ctx.New(r.Data), v.GetDisplayTemplate())
+			message, err := template(ctx, v.GetDisplayTemplate())
 			if err != nil {
 				r.ErrorMessage(err)
 			} else {
@@ -170,7 +170,7 @@ func processTemplates(ctx *context.Context, r *pkg.CheckResult) *pkg.CheckResult
 		if tpl.IsEmpty() {
 			break
 		}
-		message, err := template(ctx.New(r.Data), tpl)
+		message, err := template(ctx, tpl)
 		if err != nil {
 			r.ErrorMessage(err)
 		} else if message != "true" {

checks/sql.go

@@ -95,7 +95,21 @@ func CheckSQL(ctx *context.Context, checker SQLChecker) pkg.Results { // nolint:
 		return results.Failf("error getting connection: %v", err)
 	}
 
-	details, err := querySQL(checker.GetDriver(), connection.URL, check.GetQuery())
+	query := check.GetQuery()
+
+	if ctx.Canary.Annotations["template"] != "false" {
+		query, err = template(ctx.WithCheck(checker.GetCheck()), v1.Template{
+			Template: query,
+		})
+		if err != nil {
+			return results.ErrorMessage(err)
+		}
+		if ctx.IsDebug() {
+			ctx.Infof("query: %s", query)
+		}
+	}
+
+	details, err := querySQL(checker.GetDriver(), connection.URL, query)
 	if err != nil {
 		return results.ErrorMessage(err)
 	}

fixtures/datasources/posgres_stateful_pass.yaml

@@ -0,0 +1,25 @@
+apiVersion: canaries.flanksource.com/v1
+kind: Canary
+metadata:
+  name: postgres-succeed
+  namespace: canaries
+spec:
+  interval: 30
+  postgres:
+    - name: postgres processes new
+      url: "postgres://$(username):$(password)@postgres.canaries.svc.cluster.local:5432/postgres?sslmode=disable"
+      username:
+        value: postgresadmin
+      password:
+        value: admin123
+      query: |
+        select max(backend_start), count(*)  from pg_stat_activity WHERE backend_start >
+        {{-  if last_result.results.rows }}
+        '{{- (index last_result.results.rows 0).max }}'
+        {{- else}}
+          now() - interval '1 hour'
+        {{- end}}
+      metrics:
+        - name: postgres_process_new
+          type: counter
+          value: results.rows[0].count

fixtures/opensearch/_setup.yaml

@@ -1,7 +1,7 @@
 apiVersion: v1
 kind: Secret
 metadata:
-  name: search
+  name: opensearch
   namespace: canaries
 stringData:
   OPENSEARCH_USERNAME: admin

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/fergusstrange/embedded-postgres v1.24.0
 	github.com/flanksource/commons v1.17.1
 	github.com/flanksource/duty v1.0.205
-	github.com/flanksource/gomplate/v3 v3.20.19
+	github.com/flanksource/gomplate/v3 v3.20.22
 	github.com/flanksource/is-healthy v0.0.0-20231003215854-76c51e3a3ff7
 	github.com/flanksource/kommons v0.31.4
 	github.com/flanksource/postq v1.0.0

go.sum

@@ -820,8 +820,8 @@ github.com/flanksource/commons v1.17.1/go.mod h1:RDdQI0/QYC4GzicbDaXIvBPjWuQWKLz
 github.com/flanksource/duty v1.0.205 h1:sQq+J4TMx69NnoM4XxBcJZ8P5HM5GjY/7zcuv/IQGTo=
 github.com/flanksource/duty v1.0.205/go.mod h1:V3fgZdrBgN47lloIz7MedwD/tq4ycHI8zFOApzUpFv4=
 github.com/flanksource/gomplate/v3 v3.20.4/go.mod h1:27BNWhzzSjDed1z8YShO6W+z6G9oZXuxfNFGd/iGSdc=
-github.com/flanksource/gomplate/v3 v3.20.19 h1:xl+XMYWXtlrO6FfU+VxwjNwX4/oBK3/soOtHRvUt2us=
-github.com/flanksource/gomplate/v3 v3.20.19/go.mod h1:2GgHZ2vWmtDspJMBfUIryOuzJSwc8jU7Kw9fDLr0TMA=
+github.com/flanksource/gomplate/v3 v3.20.22 h1:DOytkLh1aND8KruydfCnu9K9oRKPeoJj2qgFqQkGrpE=
+github.com/flanksource/gomplate/v3 v3.20.22/go.mod h1:2GgHZ2vWmtDspJMBfUIryOuzJSwc8jU7Kw9fDLr0TMA=
 github.com/flanksource/is-healthy v0.0.0-20230705092916-3b4cf510c5fc/go.mod h1:4pQhmF+TnVqJroQKY8wSnSp+T18oLson6YQ2M0qPHfQ=
 github.com/flanksource/is-healthy v0.0.0-20231003215854-76c51e3a3ff7 h1:s6jf6P1pRfdvksVFjIXFRfnimvEYUR0/Mmla1EIjiRM=
 github.com/flanksource/is-healthy v0.0.0-20231003215854-76c51e3a3ff7/go.mod h1:BH5gh9JyEAuuWVP6Q5y9h43VozS0RfKyjNpM9L4v4hw=