func (s *Scaler) StreamIsActive(scaledObject *pb.ScaledObjectRef, epsServer pb.ExternalScaler_StreamIsActiveServer) error {
	ctx := epsServer.Context()
	project, p1 := scaledObject.ScalerMetadata["project"]
	service, p2 := scaledObject.ScalerMetadata["service"]
	version, p3 := scaledObject.ScalerMetadata["version"]
	if !p1 || !p2 || !p3 {
		return helpers.Logger.LogError(helpers.GetRequestID(ctx), "Invalid fields provided in keda scaler config", nil, map[string]interface{}{"project": project, "service": service, "version": version})
	}
	helpers.Logger.LogDebug(helpers.GetRequestID(ctx), "Keda stream is active called", map[string]interface{}{"project": project, "service": service, "version": version})
	ch := make(chan bool, 5)
	s.addIsActiveStream(project, service, version, ch)
	defer s.removeIsActiveStream(project, service, version)
	for {
		select {
		case <-ctx.Done():
			helpers.Logger.LogDebug(helpers.GetRequestID(ctx), "Exiting from keda is active stream", map[string]interface{}{"project": project, "service": service, "version": version})
			return nil
		case isActive := <-ch:
			if !isActive {
				return nil
			}
			helpers.Logger.LogDebug(helpers.GetRequestID(ctx), fmt.Sprintf("Streaming is active result: %v", isActive), map[string]interface{}{"project": project, "service": service, "version": version})
			if err := epsServer.Send(&pb.IsActiveResponse{Result: isActive}); err != nil {
				_ = helpers.Logger.LogError(helpers.GetRequestID(ctx), "Unable to forward is stream active resonse to keda", err, map[string]interface{}{"project": project, "service": service, "version": version})
			}
		}
	}
}
StreamIsActive is called by keda controller once to allow us to stream isActive events to it.
package scaler
import (
	"context"
	"fmt"
	"strconv"
	"github.com/spaceuptech/helpers"
	pb "github.com/spaceuptech/space-cloud/runner/modules/scaler/externalscaler"
)
func (s *Scaler) GetMetricSpec(ctx context.Context, scaledObject *pb.ScaledObjectRef) (*pb.GetMetricSpecResponse, error) {
	project, p1 := scaledObject.ScalerMetadata["project"]
	service, p2 := scaledObject.ScalerMetadata["service"]
	version, p3 := scaledObject.ScalerMetadata["version"]
	scalingMode, p4 := scaledObject.ScalerMetadata["type"]
	targetString, p5 := scaledObject.ScalerMetadata["target"]
	if !p1 || !p2 || !p3 || !p4 || !p5 {
		return nil, helpers.Logger.LogError(helpers.GetRequestID(ctx), "Invalid fields provided in keda scaler config", nil, map[string]interface{}{"project": project, "service": service, "version": version, "target": targetString})
	}
	target, err := strconv.Atoi(targetString)
	if err != nil {
		return nil, helpers.Logger.LogError(helpers.GetRequestID(ctx), "Invalid target provided in keda scaler config", nil, map[string]interface{}{"project": project, "service": service, "version": version, "target": targetString})
	}
	helpers.Logger.LogDebug(helpers.GetRequestID(ctx), fmt.Sprintf("Keda get metrics spec result: (%s, %s)", scalingMode, targetString), map[string]interface{}{"project": project, "service": service, "version": version})
	return &pb.GetMetricSpecResponse{
		MetricSpecs: []*pb.MetricSpec{{
			MetricName: scalingMode,
			TargetSize: int64(target),
		}},
	}, nil
}
func (s *Scaler) GetMetrics(ctx context.Context, metricRequest *pb.GetMetricsRequest) (*pb.GetMetricsResponse, error) {
	project, p1 := metricRequest.ScaledObjectRef.ScalerMetadata["project"]
	service, p2 := metricRequest.ScaledObjectRef.ScalerMetadata["service"]
	version, p3 := metricRequest.ScaledObjectRef.ScalerMetadata["version"]
	if !p1 || !p2 || !p3 {
		return nil, helpers.Logger.LogError(helpers.GetRequestID(ctx), "Invalid fields provided in keda scaler config", nil, map[string]interface{}{"project": project, "service": service, "version": version})
	}
	scalingMode := metricRequest.MetricName
	metric, err := s.queryPrometheus(ctx, project, service, version, scalingMode)
	if err != nil {
		return nil, helpers.Logger.LogError(helpers.GetRequestID(ctx), "Unable to fetch scaling metrics for service", err, map[string]interface{}{"project": project, "service": service, "version": version})
	}
	helpers.Logger.LogDebug(helpers.GetRequestID(ctx), fmt.Sprintf("Keda get metrics result: (%v, %v)", scalingMode, metric), map[string]interface{}{"project": project, "service": service, "version": version})
	return &pb.GetMetricsResponse{
		MetricValues: []*pb.MetricValue{{
			MetricName:  scalingMode,
			MetricValue: metric,
		}},
	}, nil
}