refactor grpc signature, cli for CRUD

pkg/alerting/shared/shared.go

@@ -15,6 +15,8 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+const OpniAlertingCortexNamespace = "opni-alerting"
+
 // Condition constants
 var ComparisonOperators = []string{"<", ">", "<=", ">=", "=", "!="}
 var KubeStates = []string{"Pending", "Running", "Succeeded", "Failed", "Unknown"}

pkg/opni/commands/admin.go

@@ -5,13 +5,18 @@ package commands
 import (
 	"encoding/json"
 	"fmt"
+	"os"
+	"sync"
 	"time"
 
 	"github.com/araddon/dateparse"
 	"github.com/samber/lo"
+	"golang.org/x/exp/slices"
+	"gopkg.in/yaml.v3"
 
 	"github.com/olebedev/when"
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/rulefmt"
 	corev1 "github.com/rancher/opni/pkg/apis/core/v1"
 	managementv1 "github.com/rancher/opni/pkg/apis/management/v1"
 	"github.com/rancher/opni/pkg/metrics/unmarshal"
@@ -62,6 +67,104 @@ func BuildCortexAdminRulesCmd() *cobra.Command {
 		Short: "Cortex admin rules",
 	}
 	cmd.AddCommand(BuildListRulesCmd())
+	cmd.AddCommand(BuildDeleteRuleGroupsCmd())
+	cmd.AddCommand(BuildLoadRuleGroupsCmd())
+	return cmd
+}
+
+func BuildDeleteRuleGroupsCmd() *cobra.Command {
+	var clusters string
+	var namespace string
+
+	cmd := &cobra.Command{
+		Use:   "delete <groupname>",
+		Short: "Delete prometheus rule groups of the given name from Cortex",
+		Args:  cobra.ExactArgs(1),
+		Run: func(cmd *cobra.Command, args []string) {
+			_, err := adminClient.DeleteRule(cmd.Context(), &cortexadmin.DeleteRuleRequest{
+				ClusterId: clusters,
+				Namespace: namespace,
+				GroupName: args[0],
+			})
+			if err != nil {
+				lg.Error(err)
+				return
+			}
+			fmt.Println("Rule Group deleted successfully")
+		},
+	}
+	cmd.Flags().StringVarP(&clusters, "cluster", "c", "", "The clusters to delete the rule from")
+	cmd.Flags().StringVarP(&namespace, "namespace", "n", "", "The namespace to delete the rule from")
+	return cmd
+}
+
+func BuildLoadRuleGroupsCmd() *cobra.Command {
+	var clusters []string
+	var namespace string
+	cmd := &cobra.Command{
+		Use:   "load <rulegroupfile>",
+		Short: "Creates/Updates prometheus rule groups into Cortex from a valid prometheus rule group file",
+		Long:  "See https://prometheus.io/docs/prometheus/latest/configuration/recording_rules/#recording-rules for more information about the expected input format",
+		Args:  cobra.ExactArgs(1),
+		Run: func(cmd *cobra.Command, args []string) {
+			//read file and validate contents
+
+			clMeta, err := mgmtClient.ListClusters(cmd.Context(), &managementv1.ListClustersRequest{})
+			if err != nil {
+				lg.Fatal(err)
+			}
+			cl := lo.Map(clMeta.Items, func(cl *corev1.Cluster, _ int) string {
+				return cl.Id
+			})
+			if len(clusters) == 0 {
+				clusters = cl
+			} else {
+				// Important to validate here !! Since cortex has no knowledge of available opni clusters,
+				// it will accept any valid yaml content and therefore could silently fail/ destroy itself
+				// by writing unsuported names to its (remote) store
+				for _, c := range clusters {
+					if !slices.Contains(cl, c) {
+						lg.Fatalf("invalid cluster id %s", c)
+					}
+				}
+			}
+			yamlContent, err := os.ReadFile(args[0])
+			if err != nil {
+				lg.Fatal(err)
+			}
+			rgs, errors := rulefmt.Parse(yamlContent)
+			if len(errors) > 0 {
+				for _, err := range errors {
+					lg.Error(err)
+				}
+				lg.Fatal("Failed to parse rule group file")
+			}
+
+			var wg sync.WaitGroup
+			for _, cl := range clusters {
+				cl := cl
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					for _, group := range rgs.Groups {
+						_, err := adminClient.LoadRules(cmd.Context(), &cortexadmin.LoadRuleRequest{
+							Namespace:   namespace,
+							ClusterId:   cl,
+							YamlContent: util.Must(yaml.Marshal(group)),
+						})
+						if err != nil {
+							lg.Errorf("Failed to load rule group :\n `%s`\n\n for cluster `%s`", string(util.Must(yaml.Marshal(group))), cl)
+						} else {
+							fmt.Printf("Successfully loaded rule group `%s` for clusterId `%s`\n", group.Name, cl)
+						}
+					}
+				}()
+			}
+			wg.Wait()
+		},
+	}
+	cmd.Flags().StringSliceVar(&clusters, "cluster", []string{}, "The clusters to apply the rule to (default=all)")
+	cmd.Flags().StringVar(&namespace, "namespace", "", "namespace is a cortex identifier to help organize rules (default=\"default\")")
 	return cmd
 }
 

pkg/test/util.go

@@ -72,8 +72,9 @@ func ExpectRuleGroupToExist(
 	maxTimeout time.Duration,
 ) {
 	Eventually(func() error {
-		_, err := client.GetRule(ctx, &cortexadmin.RuleRequest{
+		_, err := client.GetRule(ctx, &cortexadmin.GetRuleRequest{
 			ClusterId: tenant,
+			Namespace: "test",
 			GroupName: groupName,
 		})
 		if err != nil {

plugins/alerting/pkg/alerting/condition_handlers.go

@@ -107,8 +107,9 @@ func deleteCondition(p *Plugin, lg *zap.SugaredLogger, ctx context.Context, req
 		return nil
 	}
 	if r, _ := handleSwitchCortexRules(req.AlertType); r != nil {
-		_, err := p.adminClient.Get().DeleteRule(ctx, &cortexadmin.RuleRequest{
+		_, err := p.adminClient.Get().DeleteRule(ctx, &cortexadmin.DeleteRuleRequest{
 			ClusterId: r.Id,
+			Namespace: shared.OpniAlertingCortexNamespace,
 			GroupName: CortexRuleIdFromUuid(id),
 		})
 		return err
@@ -197,9 +198,10 @@ func (p *Plugin) handleKubeAlertCreation(ctx context.Context, k *alertingv1.Aler
 		return err
 	}
 	p.Logger.With("Expr", "kube-state").Debugf("%s", string(out))
-	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.PostRuleRequest{
+	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.LoadRuleRequest{
 		ClusterId:   k.GetClusterId(),
-		YamlContent: string(out),
+		Namespace:   shared.OpniAlertingCortexNamespace,
+		YamlContent: out,
 	})
 	if err != nil {
 		return err
@@ -234,9 +236,10 @@ func (p *Plugin) handleCpuSaturationAlertCreation(
 	}
 	p.Logger.With("Expr", "cpu").Debugf("%s", string(out))
 
-	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.PostRuleRequest{
+	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.LoadRuleRequest{
 		ClusterId:   c.ClusterId.GetId(),
-		YamlContent: string(out),
+		Namespace:   shared.OpniAlertingCortexNamespace,
+		YamlContent: out,
 	})
 	return err
 }
@@ -263,9 +266,10 @@ func (p *Plugin) handleMemorySaturationAlertCreation(ctx context.Context, m *ale
 		return err
 	}
 	p.Logger.With("Expr", "mem").Debugf("%s", string(out))
-	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.PostRuleRequest{
+	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.LoadRuleRequest{
 		ClusterId:   m.ClusterId.GetId(),
-		YamlContent: string(out),
+		Namespace:   shared.OpniAlertingCortexNamespace,
+		YamlContent: out,
 	})
 	return err
 }
@@ -291,9 +295,10 @@ func (p *Plugin) handleFsSaturationAlertCreation(ctx context.Context, fs *alerti
 		return err
 	}
 	p.Logger.With("Expr", "fs").Debugf("%s", string(out))
-	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.PostRuleRequest{
+	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.LoadRuleRequest{
 		ClusterId:   fs.ClusterId.GetId(),
-		YamlContent: string(out),
+		Namespace:   shared.OpniAlertingCortexNamespace,
+		YamlContent: out,
 	})
 	return err
 }
@@ -317,9 +322,9 @@ func (p *Plugin) handlePrometheusQueryAlertCreation(ctx context.Context, q *aler
 		return err
 	}
 	p.Logger.With("Expr", "user-query").Debugf("%s", string(out))
-	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.PostRuleRequest{
+	_, err = p.adminClient.Get().LoadRules(ctx, &cortexadmin.LoadRuleRequest{
 		ClusterId:   q.ClusterId.GetId(),
-		YamlContent: string(out),
+		YamlContent: out,
 	})
 
 	return err