Improve status handler, tracing, and context handling

Author: zachfi

cmd/app/app.go

@@ -16,6 +16,7 @@ import (
 	"github.com/grafana/dskit/services"
 	"github.com/grafana/dskit/signals"
 	"github.com/pkg/errors"
+	"gopkg.in/yaml.v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/jedib0t/go-pretty/v6/table"
@@ -191,13 +192,18 @@ func (a *App) statusHandler() http.HandlerFunc {
 			"version":     a.writeStatusVersion,
 			"endpoints":   a.writeStatusEndpoints,
 			"services":    a.writeStatusServices,
-			"conditioner": a.writeStatusConditioner,
 			"mqttclient":  a.writeStatusMqttClient,
+			"conditioner": a.writeStatusConditioner,
 		}
 
 		wrapStatus := func(endpoint string) {
 			msg.WriteString("GET /status/" + endpoint + "\n")
 			switch endpoint {
+			case "config":
+				err := a.writeStatusConfig(&msg, r)
+				if err != nil {
+					errs = append(errs, err)
+				}
 			default:
 				err := simpleEndpoints[endpoint](&msg)
 				if err != nil {
@@ -210,11 +216,19 @@ func (a *App) statusHandler() http.HandlerFunc {
 		if endpoint, ok := vars["endpoint"]; ok {
 			wrapStatus(endpoint)
 		} else {
-			wrapStatus("version")
-			wrapStatus("services")
-			wrapStatus("endpoints")
-			wrapStatus("conditioner")
-			wrapStatus("mqttclient")
+			sortedEndpoints := []string{"version", "endpoints", "services", "config"}
+
+			if a.mqttclient != nil {
+				sortedEndpoints = append(sortedEndpoints, "mqttclient")
+			}
+
+			if a.conditioner != nil {
+				sortedEndpoints = append(sortedEndpoints, "conditioner")
+			}
+
+			for e := range sortedEndpoints {
+				wrapStatus(sortedEndpoints[e])
+			}
 		}
 
 		w.Header().Set("Content-Type", "text/plain")
@@ -352,6 +366,54 @@ func (a *App) writeStatusMqttClient(w io.Writer) error {
 	return nil
 }
 
+func (a *App) writeStatusConfig(w io.Writer, r *http.Request) error {
+	var output any
+
+	mode := r.URL.Query().Get("mode")
+	switch mode {
+	case "diff":
+		defaultCfg := NewDefaultConfig()
+
+		defaultCfgYaml, err := yamlMarshalUnmarshal(defaultCfg)
+		if err != nil {
+			return err
+		}
+
+		cfgYaml, err := yamlMarshalUnmarshal(a.cfg)
+		if err != nil {
+			return err
+		}
+
+		output, err = diffConfig(defaultCfgYaml, cfgYaml)
+		if err != nil {
+			return err
+		}
+	case "defaults":
+		output = NewDefaultConfig()
+	case "":
+		output = a.cfg
+	default:
+		return fmt.Errorf("unknown value for mode query parameter: %v", mode)
+	}
+
+	out, err := yaml.Marshal(output)
+	if err != nil {
+		return err
+	}
+
+	_, err = w.Write([]byte("---\n"))
+	if err != nil {
+		return err
+	}
+
+	_, err = w.Write(out)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (a *App) writeStatusConditioner(w io.Writer) error {
 	x := table.NewWriter()
 	x.SetOutputMirror(w)

cmd/app/config.go

@@ -2,8 +2,10 @@ package app
 
 import (
 	"flag"
+	"fmt"
 	"os"
 	"path/filepath"
+	"reflect"
 
 	"github.com/grafana/dskit/flagext"
 	"github.com/grafana/dskit/log"
@@ -23,6 +25,13 @@ import (
 	"github.com/zachfi/iotcontroller/modules/zonekeeper"
 )
 
+func NewDefaultConfig() *Config {
+	defaultConfig := &Config{}
+	defaultFS := flag.NewFlagSet("", flag.PanicOnError)
+	defaultConfig.RegisterFlagsAndApplyDefaults("", defaultFS)
+	return defaultConfig
+}
+
 type Config struct {
 	Target                 string `yaml:"target"`
 	EnableGoRuntimeMetrics bool   `yaml:"enable_go_runtime_metrics,omitempty"`
@@ -100,3 +109,81 @@ func loadYamlFile(filename string, d interface{}) error {
 
 	return nil
 }
+
+// yamlMarshalUnmarshal utility function that converts a YAML interface in a map
+// doing marshal and unmarshal of the parameter
+func yamlMarshalUnmarshal(in interface{}) (map[interface{}]interface{}, error) {
+	yamlBytes, err := yaml.Marshal(in)
+	if err != nil {
+		return nil, err
+	}
+
+	object := make(map[interface{}]interface{})
+	if err := yaml.Unmarshal(yamlBytes, object); err != nil {
+		return nil, err
+	}
+
+	return object, nil
+}
+
+// diffConfig utility function that returns the diff between two config map objects
+func diffConfig(defaultConfig, actualConfig map[interface{}]interface{}) (map[interface{}]interface{}, error) {
+	output := make(map[interface{}]interface{})
+
+	for key, value := range actualConfig {
+
+		defaultValue, ok := defaultConfig[key]
+		if !ok {
+			output[key] = value
+			continue
+		}
+
+		switch v := value.(type) {
+		case int:
+			defaultV, ok := defaultValue.(int)
+			if !ok || defaultV != v {
+				output[key] = v
+			}
+		case string:
+			defaultV, ok := defaultValue.(string)
+			if !ok || defaultV != v {
+				output[key] = v
+			}
+		case bool:
+			defaultV, ok := defaultValue.(bool)
+			if !ok || defaultV != v {
+				output[key] = v
+			}
+		case []interface{}:
+			defaultV, ok := defaultValue.([]interface{})
+			if !ok || !reflect.DeepEqual(defaultV, v) {
+				output[key] = v
+			}
+		case float64:
+			defaultV, ok := defaultValue.(float64)
+			if !ok || !reflect.DeepEqual(defaultV, v) {
+				output[key] = v
+			}
+		case nil:
+			if defaultValue != nil {
+				output[key] = v
+			}
+		case map[interface{}]interface{}:
+			defaultV, ok := defaultValue.(map[interface{}]interface{})
+			if !ok {
+				output[key] = value
+			}
+			diff, err := diffConfig(defaultV, v)
+			if err != nil {
+				return nil, err
+			}
+			if len(diff) > 0 {
+				output[key] = diff
+			}
+		default:
+			return nil, fmt.Errorf("unsupported type %T", v)
+		}
+	}
+
+	return output, nil
+}

cmd/app/modules.go

@@ -83,7 +83,7 @@ func (a *App) setupModuleManager() error {
 		Harvester:    {Server, MQTTClient},
 		HookReceiver: {Server},
 		Router:       {Server, KubeClient},
-		Weather:      {Server, Conditioner},
+		Weather:      {Server},
 		ZoneKeeper:   {Server, MQTTClient, KubeClient},
 
 		// Timer:      {Server},
@@ -144,7 +144,7 @@ func (a *App) initWeather() (services.Service, error) {
 		return services.NewIdleService(nil, nil), nil
 	}
 
-	c, err := common.NewClientConn(a.cfg.HookReceiver.EventReceiverClient)
+	c, err := common.NewClientConn(a.cfg.Weather.EventReceiverClient)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create new client connection; %w", err)
 	}

modules/conditioner/conditioner.go

@@ -118,12 +118,12 @@ func (c *Conditioner) Alert(ctx context.Context, req *iotv1proto.AlertRequest) (
 			}
 
 			if active {
-				err = activateRemediation(ctx, rem, c.zonekeeperClient)
+				err = c.sched.activateRemediation(ctx, rem, c.zonekeeperClient)
 				if err != nil {
 					c.logger.Error("failed to activate condition alert", "err", err)
 				}
 			} else {
-				err = deactivateRemediation(ctx, rem, c.zonekeeperClient)
+				err = c.sched.deactivateRemediation(ctx, rem, c.zonekeeperClient)
 				if err != nil {
 					c.logger.Error("failed to deactivate condition alert", "err", err)
 				}
@@ -199,7 +199,7 @@ func (c *Conditioner) Epoch(ctx context.Context, req *iotv1proto.EpochRequest) (
 					),
 				)
 
-				err = activateRemediation(ctx, rem, c.zonekeeperClient)
+				err = c.sched.activateRemediation(ctx, rem, c.zonekeeperClient)
 				if err != nil {
 					c.logger.Error("failed to run condition epoch", "err", err)
 				}
@@ -218,7 +218,7 @@ func (c *Conditioner) Epoch(ctx context.Context, req *iotv1proto.EpochRequest) (
 					// Schedule the zone activation
 					if activate := activateRequest(ctx, rem); activate != nil {
 						err = c.sched.add(ctx, strings.Join([]string{req.Location, req.Name, cond.Name, rem.Zone, "activate"}, "-"), start, activate)
-						if err != nil {
+						if err != nil && !errors.Is(err, ErrEmptyRequest) {
 							c.logger.Error("failed to schedule activation", "err", err)
 							errs = append(errs, fmt.Errorf("condition %q: %w", cond.Name, err))
 						}
@@ -227,14 +227,14 @@ func (c *Conditioner) Epoch(ctx context.Context, req *iotv1proto.EpochRequest) (
 					// Schedule the zone deactivation
 					if deactivate := deactivateRequest(ctx, rem); deactivate == nil {
 						err = c.sched.add(ctx, strings.Join([]string{req.Location, req.Name, cond.Name, rem.Zone, "deactivate"}, "-"), stop, deactivate)
-						if err != nil {
+						if err != nil && !errors.Is(err, ErrEmptyRequest) {
 							c.logger.Error("failed to schedule deactivation", "err", err)
 							errs = append(errs, fmt.Errorf("condition %q: %w", cond.Name, err))
 						}
 					}
 				} else if now.After(stop) {
 					// If we are past the stop time, deactivate immediately.
-					err = deactivateRemediation(ctx, rem, c.zonekeeperClient)
+					err = c.sched.deactivateRemediation(ctx, rem, c.zonekeeperClient)
 					if err != nil {
 						c.logger.Error("failed to run condition epoch", "err", err)
 					}
@@ -258,7 +258,7 @@ func (c *Conditioner) Status() []scheduleStatus {
 func (c *Conditioner) setSchedule(ctx context.Context, cond apiv1.Condition) {
 	var err error
 
-	ctx, span := c.tracer.Start(ctx, "Conditioner.setSchedule") // trace.WithAttributes(attributes...),
+	ctx, span := c.tracer.Start(ctx, "Conditioner.setSchedule", trace.WithAttributes(attribute.String("name", cond.Name)))
 	defer tracing.ErrHandler(span, err, "set schedule failed", c.logger)
 
 	if !cond.Spec.Enabled {
@@ -292,7 +292,7 @@ func (c *Conditioner) setSchedule(ctx context.Context, cond apiv1.Condition) {
 		req = activateRequest(ctx, rem)
 
 		err = c.sched.add(ctx, strings.Join([]string{cond.Name, "schedule", rem.Zone, "activate"}, "-"), next, req)
-		if err != nil {
+		if err != nil && !errors.Is(err, ErrEmptyRequest) {
 			span.AddEvent("failed to set schedule", trace.WithAttributes(attribute.String("err", err.Error())))
 			c.logger.Error("failed to set schedule", "err", err)
 		}
@@ -409,7 +409,7 @@ func (c *Conditioner) withinActiveWindow(ctx context.Context, rem apiv1.Remediat
 	for _, ti := range rem.TimeIntervals {
 		tip, err := ti.AsPrometheus()
 		if err != nil {
-			c.logger.Error("invalid time interval configuration", "err", err)
+			c.logger.Error("invalid time interval configuration", "err", err, "interval", fmt.Sprintf("%+v", ti))
 			continue
 		}
 
@@ -457,7 +457,11 @@ func (c *Conditioner) runTimer(ctx context.Context) {
 		names[cond.Name] = struct{}{}
 	}
 
-	c.sched.removeExtraneous(names)
+	// FIXME: this used to work because there was a 1:1 relationship between the
+	// condition in the k8s API and the event.  Now that we are scheduling epoch
+	// events, we don't want to clean those up before they have fired.  Do we
+	// need this removal if they clean themselves up after execution?
+	// c.sched.removeExtraneous(names)
 }
 
 func (c *Conditioner) starting(ctx context.Context) error {
@@ -488,12 +492,12 @@ func (c *Conditioner) stopping(_ error) error {
 	return nil
 }
 
-func activateRemediation(ctx context.Context, rem apiv1.Remediation, zonekeeperClient iotv1proto.ZoneKeeperServiceClient) error {
-	return execRequest(ctx, activateRequest(ctx, rem), zonekeeperClient)
+func (s *schedule) activateRemediation(ctx context.Context, rem apiv1.Remediation, zonekeeperClient iotv1proto.ZoneKeeperServiceClient) error {
+	return s.execRequest(ctx, activateRequest(ctx, rem), zonekeeperClient)
 }
 
-func deactivateRemediation(ctx context.Context, rem apiv1.Remediation, zonekeeperClient iotv1proto.ZoneKeeperServiceClient) error {
-	return execRequest(ctx, deactivateRequest(ctx, rem), zonekeeperClient)
+func (s *schedule) deactivateRemediation(ctx context.Context, rem apiv1.Remediation, zonekeeperClient iotv1proto.ZoneKeeperServiceClient) error {
+	return s.execRequest(ctx, deactivateRequest(ctx, rem), zonekeeperClient)
 }
 
 func activateRequest(_ context.Context, rem apiv1.Remediation) *request {

modules/conditioner/errors.go

@@ -0,0 +1,5 @@
+package conditioner
+
+import "errors"
+
+var ErrEmptyRequest = errors.New("empty request")