KEDA External Scaler

작동 방식

KEDA는 KEDA operator (0 ↔ 1 스케일링)와 Kubernetes HPA (1 ↔ N 스케일링)를 활용한 2단계 접근 방식을 사용하여 scaleTargetRef (e.g., Deployment)를 스케일링합니다.

다음은 단계별 메커니즘입니다.

1. 0에서 1로 스케일링 (활성화); KEDA operator는 scaleTargetRef를 "깨우는" 역할을 합니다.

   • Loop: KEDA는 정기적으로 외부 스케일러의 IsActive RPC 메서드를 폴링합니다.
   • Logic:
     • IsActive가 true를 반환하면: KEDA는 scaleTargetRef를 minReplicaCount (기본값 1)로 스케일링합니다.
     • IsActive가 (쿨다운 기간 동안) false를 반환하면: KEDA는 scaleTargetRef를 0으로 스케일링합니다.

2. 1에서 N으로 스케일링 (HPA); scaleTargetRef가 활성화되면 (replicas > 0), KEDA는 스케일링을 처리하기 위해 표준 Kubernetes HPA 리소스를 생성합니다.

   • Setup: KEDA는 HPA 규칙을 정의하기 위해 스케일러의 GetMetricSpec RPC를 호출합니다 (e.g., TargetSize: 10).
   • Loop: Kubernetes HPA 컨트롤러는 KEDA metrics server를 폴링합니다.
   • Data flow:
     1. HPA는 KEDA metrics server에 질문합니다.
     2. KEDA metrics server는 스케일러의 GetMetrics RPC를 호출합니다.
     3. 스케일러는 현재 값을 반환합니다 (e.g., 50).
     4. HPA는 표준 공식을 사용하여 desiredReplicas를 계산합니다. Kubernetes HPA # Algorithm

$$\text{desiredReplicas} = \lceil \text{currentReplicas} \times \frac{\text{currentMetricValue}}{\text{desiredMetricValue}} \rceil$$

• desiredMetricValue is TargetSize
• If metricType is AverageValue: $$\text{currentMetricValue} = \frac{\text{metricValue}}{\text{currentReplicas}}$$
• If metricType is Value: $$\text{currentMetricValue} = \text{metricValue}$$

Scaler 서비스 구현

References

• KEDA / Concepts / External Scalers
• gRPC

아래와 같은 구조의 프로젝트를 생성합니다.

<externalscaler>
├── cmd/
│   └── scaler/
│       └── main.go
├── internal/
│   ├── application/
│   │   └── externalscaler.go
│   └── pkg/
│       └── externalscaler/
│           ├── externalscaler.pb.go
│           ├── externalscaler.proto
│           └── externalscaler_grpc.pb.go
├── go.mod
├── go.sum
├── Makefile
└── README.md

curl https://raw.githubusercontent.com/kedacore/keda/main/pkg/scalers/externalscaler/externalscaler.proto \
    --create-dirs \
    -o internal/pkg/externalscaler/externalscaler.proto

protoc \
    --go_out=. \
    --go_opt=paths=source_relative \
    --go-grpc_out=. \
    --go-grpc_opt=paths=source_relative \
    internal/pkg/externalscaler/externalscaler.proto

internal/application/externalscaler.go

package application

import (
	"context"

	pb "github.com/hhk7734/externalscaler-test/internal/pkg/externalscaler"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type ExternalScaler struct {
	pb.UnimplementedExternalScalerServer
}

// IsActive는 대상이 활성화되어야 하는지 여부를 반환합니다. false를 반환하면 대상의 replicas를 0으로
// 설정합니다.
func (e *ExternalScaler) IsActive(ctx context.Context, scaledObject *pb.ScaledObjectRef) (*pb.IsActiveResponse, error) {
	// value := scaledObject.ScalerMetadata["<customKey>"]

	return &pb.IsActiveResponse{Result: true}, nil
}

// GetMetricSpec은 대상이 스캐일링되기 위한 메트릭의 기준값을 반환합니다.
func (e *ExternalScaler) GetMetricSpec(ctx context.Context, scaledObject *pb.ScaledObjectRef) (*pb.GetMetricSpecResponse, error) {
	// value := scaledObject.ScalerMetadata["<customKey>"]

	return &pb.GetMetricSpecResponse{
		MetricSpecs: []*pb.MetricSpec{
			{
				MetricName: "custom-metric",
				TargetSize: 10,
			},
		},
	}, nil
}

// GetMetrics은 메트릭 값을 반환합니다.
func (e *ExternalScaler) GetMetrics(ctx context.Context, request *pb.GetMetricsRequest) (*pb.GetMetricsResponse, error) {
	// value := request.ScaledObjectRef.ScalerMetadata["<customKey>"]

	return &pb.GetMetricsResponse{
		MetricValues: []*pb.MetricValue{
			{
				MetricName:  "custom-metric",
				MetricValue: 1,
			},
		},
	}, nil
}

// StreamIsActive은 spec.triggers.type: external-push인 경우에만 사용됩니다.
func (e *ExternalScaler) StreamIsActive(scaledObject *pb.ScaledObjectRef, stream pb.ExternalScaler_StreamIsActiveServer) error {
	// value := scaledObject.ScalerMetadata["<customKey>"]

	// for {
	// 	select {
	// 	case <-stream.Context().Done():
	// 		return nil
	// 	case <-time.Tick(30 * time.Second):
	// 		if err := stream.Send(&pb.IsActiveResponse{Result: true}); err != nil {
	// 			return err
	// 		}
	// 	}
	// }

	return status.Error(codes.Unavailable, "external-push is not supported")
}

cmd/scaler/main.go

package main

import (
	"net"

	"github.com/hhk7734/externalscaler-test/internal/application"
	pb "github.com/hhk7734/externalscaler-test/internal/pkg/externalscaler"
	"google.golang.org/grpc"
)

func main() {
	grpcServer := grpc.NewServer()
	lis, _ := net.Listen("tcp", ":6000")
	pb.RegisterExternalScalerServer(grpcServer, &application.ExternalScaler{})

	if err := grpcServer.Serve(lis); err != nil {
	}
}

CRD에 External scaler 설정하기

spec:
  triggers:
    - type: external
      metricType: AverageValue
      metadata:
        scalerAddress: <host>:<port>
        # External scaler에 전달할 메타데이터
        # <customKey>: <value>