diff --git a/flyteidl/clients/go/admin/mocks/AdminServiceClient.go b/flyteidl/clients/go/admin/mocks/AdminServiceClient.go
index 0854c88eb68..84b88e1d1e1 100644
--- a/flyteidl/clients/go/admin/mocks/AdminServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/AdminServiceClient.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	grpc "google.golang.org/grpc"
-
 	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // AdminServiceClient is an autogenerated mock type for the AdminServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/AdminServiceServer.go b/flyteidl/clients/go/admin/mocks/AdminServiceServer.go
index afc69cd991a..a05b851f71a 100644
--- a/flyteidl/clients/go/admin/mocks/AdminServiceServer.go
+++ b/flyteidl/clients/go/admin/mocks/AdminServiceServer.go
@@ -6,7 +6,6 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
 	mock "github.com/stretchr/testify/mock"
 )
 
diff --git a/flyteidl/clients/go/admin/mocks/AgentMetadataServiceClient.go b/flyteidl/clients/go/admin/mocks/AgentMetadataServiceClient.go
index 0fef13afab4..62acbd54c53 100644
--- a/flyteidl/clients/go/admin/mocks/AgentMetadataServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/AgentMetadataServiceClient.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	grpc "google.golang.org/grpc"
-
 	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // AgentMetadataServiceClient is an autogenerated mock type for the AgentMetadataServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/AgentMetadataServiceServer.go b/flyteidl/clients/go/admin/mocks/AgentMetadataServiceServer.go
index 71e698ef2aa..85230f7db83 100644
--- a/flyteidl/clients/go/admin/mocks/AgentMetadataServiceServer.go
+++ b/flyteidl/clients/go/admin/mocks/AgentMetadataServiceServer.go
@@ -6,7 +6,6 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
 	mock "github.com/stretchr/testify/mock"
 )
 
diff --git a/flyteidl/clients/go/admin/mocks/AsyncAgentServiceClient.go b/flyteidl/clients/go/admin/mocks/AsyncAgentServiceClient.go
index c537228482a..ffc5ad62d9d 100644
--- a/flyteidl/clients/go/admin/mocks/AsyncAgentServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/AsyncAgentServiceClient.go
@@ -6,12 +6,9 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	grpc "google.golang.org/grpc"
-
-	mock "github.com/stretchr/testify/mock"
-
 	service "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/service"
+	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // AsyncAgentServiceClient is an autogenerated mock type for the AsyncAgentServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/AsyncAgentServiceServer.go b/flyteidl/clients/go/admin/mocks/AsyncAgentServiceServer.go
index 5c489efc9f3..52a48436f74 100644
--- a/flyteidl/clients/go/admin/mocks/AsyncAgentServiceServer.go
+++ b/flyteidl/clients/go/admin/mocks/AsyncAgentServiceServer.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	mock "github.com/stretchr/testify/mock"
-
 	service "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/service"
+	mock "github.com/stretchr/testify/mock"
 )
 
 // AsyncAgentServiceServer is an autogenerated mock type for the AsyncAgentServiceServer type
diff --git a/flyteidl/clients/go/admin/mocks/AsyncAgentService_GetTaskLogsClient.go b/flyteidl/clients/go/admin/mocks/AsyncAgentService_GetTaskLogsClient.go
index fe19dbc1593..a684f728c75 100644
--- a/flyteidl/clients/go/admin/mocks/AsyncAgentService_GetTaskLogsClient.go
+++ b/flyteidl/clients/go/admin/mocks/AsyncAgentService_GetTaskLogsClient.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	metadata "google.golang.org/grpc/metadata"
-
 	mock "github.com/stretchr/testify/mock"
+	metadata "google.golang.org/grpc/metadata"
 )
 
 // AsyncAgentService_GetTaskLogsClient is an autogenerated mock type for the AsyncAgentService_GetTaskLogsClient type
diff --git a/flyteidl/clients/go/admin/mocks/AsyncAgentService_GetTaskLogsServer.go b/flyteidl/clients/go/admin/mocks/AsyncAgentService_GetTaskLogsServer.go
index b6927e4b2e9..6064c0a90dc 100644
--- a/flyteidl/clients/go/admin/mocks/AsyncAgentService_GetTaskLogsServer.go
+++ b/flyteidl/clients/go/admin/mocks/AsyncAgentService_GetTaskLogsServer.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	metadata "google.golang.org/grpc/metadata"
-
 	mock "github.com/stretchr/testify/mock"
+	metadata "google.golang.org/grpc/metadata"
 )
 
 // AsyncAgentService_GetTaskLogsServer is an autogenerated mock type for the AsyncAgentService_GetTaskLogsServer type
diff --git a/flyteidl/clients/go/admin/mocks/AuthMetadataServiceClient.go b/flyteidl/clients/go/admin/mocks/AuthMetadataServiceClient.go
index 595b5fa04c0..1c9288fac42 100644
--- a/flyteidl/clients/go/admin/mocks/AuthMetadataServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/AuthMetadataServiceClient.go
@@ -5,11 +5,9 @@ package mocks
 import (
 	context "context"
 
-	grpc "google.golang.org/grpc"
-
-	mock "github.com/stretchr/testify/mock"
-
 	service "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/service"
+	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // AuthMetadataServiceClient is an autogenerated mock type for the AuthMetadataServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/DataProxyServiceClient.go b/flyteidl/clients/go/admin/mocks/DataProxyServiceClient.go
index b2593fcf7e4..6dc0dacd316 100644
--- a/flyteidl/clients/go/admin/mocks/DataProxyServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/DataProxyServiceClient.go
@@ -5,11 +5,9 @@ package mocks
 import (
 	context "context"
 
-	grpc "google.golang.org/grpc"
-
-	mock "github.com/stretchr/testify/mock"
-
 	service "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/service"
+	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // DataProxyServiceClient is an autogenerated mock type for the DataProxyServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/ExternalPluginServiceClient.go b/flyteidl/clients/go/admin/mocks/ExternalPluginServiceClient.go
index 865bf9bcd81..6322b9fd7ff 100644
--- a/flyteidl/clients/go/admin/mocks/ExternalPluginServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/ExternalPluginServiceClient.go
@@ -5,11 +5,9 @@ package mocks
 import (
 	context "context"
 
-	grpc "google.golang.org/grpc"
-
-	mock "github.com/stretchr/testify/mock"
-
 	service "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/service"
+	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // ExternalPluginServiceClient is an autogenerated mock type for the ExternalPluginServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/IdentityServiceClient.go b/flyteidl/clients/go/admin/mocks/IdentityServiceClient.go
index 961bac5b2bc..efecbb87aff 100644
--- a/flyteidl/clients/go/admin/mocks/IdentityServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/IdentityServiceClient.go
@@ -5,11 +5,9 @@ package mocks
 import (
 	context "context"
 
-	grpc "google.golang.org/grpc"
-
-	mock "github.com/stretchr/testify/mock"
-
 	service "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/service"
+	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // IdentityServiceClient is an autogenerated mock type for the IdentityServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/SignalServiceClient.go b/flyteidl/clients/go/admin/mocks/SignalServiceClient.go
index 11cd4e9f9f1..04bc1783b0c 100644
--- a/flyteidl/clients/go/admin/mocks/SignalServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/SignalServiceClient.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	grpc "google.golang.org/grpc"
-
 	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // SignalServiceClient is an autogenerated mock type for the SignalServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/SignalServiceServer.go b/flyteidl/clients/go/admin/mocks/SignalServiceServer.go
index 7b04736a5bd..6a13d1be2b6 100644
--- a/flyteidl/clients/go/admin/mocks/SignalServiceServer.go
+++ b/flyteidl/clients/go/admin/mocks/SignalServiceServer.go
@@ -6,7 +6,6 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
 	mock "github.com/stretchr/testify/mock"
 )
 
diff --git a/flyteidl/clients/go/admin/mocks/SyncAgentServiceClient.go b/flyteidl/clients/go/admin/mocks/SyncAgentServiceClient.go
index 6c39f1fed60..b5efa0e1de2 100644
--- a/flyteidl/clients/go/admin/mocks/SyncAgentServiceClient.go
+++ b/flyteidl/clients/go/admin/mocks/SyncAgentServiceClient.go
@@ -5,11 +5,9 @@ package mocks
 import (
 	context "context"
 
-	grpc "google.golang.org/grpc"
-
-	mock "github.com/stretchr/testify/mock"
-
 	service "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/service"
+	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // SyncAgentServiceClient is an autogenerated mock type for the SyncAgentServiceClient type
diff --git a/flyteidl/clients/go/admin/mocks/SyncAgentService_ExecuteTaskSyncClient.go b/flyteidl/clients/go/admin/mocks/SyncAgentService_ExecuteTaskSyncClient.go
index 6d161bb9538..932c4e1c4e4 100644
--- a/flyteidl/clients/go/admin/mocks/SyncAgentService_ExecuteTaskSyncClient.go
+++ b/flyteidl/clients/go/admin/mocks/SyncAgentService_ExecuteTaskSyncClient.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	metadata "google.golang.org/grpc/metadata"
-
 	mock "github.com/stretchr/testify/mock"
+	metadata "google.golang.org/grpc/metadata"
 )
 
 // SyncAgentService_ExecuteTaskSyncClient is an autogenerated mock type for the SyncAgentService_ExecuteTaskSyncClient type
diff --git a/flyteidl/clients/go/admin/mocks/SyncAgentService_ExecuteTaskSyncServer.go b/flyteidl/clients/go/admin/mocks/SyncAgentService_ExecuteTaskSyncServer.go
index 61b4691a72a..802141dc1d4 100644
--- a/flyteidl/clients/go/admin/mocks/SyncAgentService_ExecuteTaskSyncServer.go
+++ b/flyteidl/clients/go/admin/mocks/SyncAgentService_ExecuteTaskSyncServer.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	admin "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/admin"
-
-	metadata "google.golang.org/grpc/metadata"
-
 	mock "github.com/stretchr/testify/mock"
+	metadata "google.golang.org/grpc/metadata"
 )
 
 // SyncAgentService_ExecuteTaskSyncServer is an autogenerated mock type for the SyncAgentService_ExecuteTaskSyncServer type
diff --git a/flyteidl/clients/go/cacheservice/mocks/CacheServiceClient.go b/flyteidl/clients/go/cacheservice/mocks/CacheServiceClient.go
index 90a99c59073..05f64d9ccec 100644
--- a/flyteidl/clients/go/cacheservice/mocks/CacheServiceClient.go
+++ b/flyteidl/clients/go/cacheservice/mocks/CacheServiceClient.go
@@ -6,10 +6,8 @@ import (
 	context "context"
 
 	cacheservice "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/cacheservice"
-
-	grpc "google.golang.org/grpc"
-
 	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // CacheServiceClient is an autogenerated mock type for the CacheServiceClient type
diff --git a/flyteidl/clients/go/datacatalog/mocks/DataCatalogClient.go b/flyteidl/clients/go/datacatalog/mocks/DataCatalogClient.go
index 19e296eb1d9..d21d74ba42e 100644
--- a/flyteidl/clients/go/datacatalog/mocks/DataCatalogClient.go
+++ b/flyteidl/clients/go/datacatalog/mocks/DataCatalogClient.go
@@ -6,9 +6,8 @@ import (
 	context "context"
 
 	datacatalog "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/datacatalog"
-	grpc "google.golang.org/grpc"
-
 	mock "github.com/stretchr/testify/mock"
+	grpc "google.golang.org/grpc"
 )
 
 // DataCatalogClient is an autogenerated mock type for the DataCatalogClient type
diff --git a/flyteidl/gen/pb-es/flyteidl/plugins/ray_pb.ts b/flyteidl/gen/pb-es/flyteidl/plugins/ray_pb.ts
index 0c0a33e0a1c..0fbba912515 100644
--- a/flyteidl/gen/pb-es/flyteidl/plugins/ray_pb.ts
+++ b/flyteidl/gen/pb-es/flyteidl/plugins/ray_pb.ts
@@ -5,7 +5,8 @@
 
 import type { BinaryReadOptions, FieldList, JsonReadOptions, JsonValue, PartialMessage, PlainMessage } from "@bufbuild/protobuf";
 import { Message, proto3 } from "@bufbuild/protobuf";
-import { K8sPod } from "../core/tasks_pb.js";
+import { KeyValuePair } from "../core/literals_pb.js";
+import { K8sPod, Resources } from "../core/tasks_pb.js";
 
 /**
  * RayJobSpec defines the desired state of RayJob
@@ -83,6 +84,105 @@ export class RayJob extends Message<RayJob> {
   }
 }
 
+/**
+ * @generated from message flyteidl.plugins.AutoscalerOptions
+ */
+export class AutoscalerOptions extends Message<AutoscalerOptions> {
+  /**
+   * @generated from field: flyteidl.plugins.AutoscalerOptions.UpscalingMode upscaling_mode = 1;
+   */
+  upscalingMode = AutoscalerOptions_UpscalingMode.UNSPECIFIED;
+
+  /**
+   * @generated from field: int32 idle_timeout_seconds = 2;
+   */
+  idleTimeoutSeconds = 0;
+
+  /**
+   * autoscaler sidecar env vars
+   *
+   * @generated from field: repeated flyteidl.core.KeyValuePair env = 3;
+   */
+  env: KeyValuePair[] = [];
+
+  /**
+   * custom autoscaler image
+   *
+   * @generated from field: string image = 4;
+   */
+  image = "";
+
+  /**
+   * autoscaler container resources
+   *
+   * @generated from field: flyteidl.core.Resources resources = 5;
+   */
+  resources?: Resources;
+
+  constructor(data?: PartialMessage<AutoscalerOptions>) {
+    super();
+    proto3.util.initPartial(data, this);
+  }
+
+  static readonly runtime: typeof proto3 = proto3;
+  static readonly typeName = "flyteidl.plugins.AutoscalerOptions";
+  static readonly fields: FieldList = proto3.util.newFieldList(() => [
+    { no: 1, name: "upscaling_mode", kind: "enum", T: proto3.getEnumType(AutoscalerOptions_UpscalingMode) },
+    { no: 2, name: "idle_timeout_seconds", kind: "scalar", T: 5 /* ScalarType.INT32 */ },
+    { no: 3, name: "env", kind: "message", T: KeyValuePair, repeated: true },
+    { no: 4, name: "image", kind: "scalar", T: 9 /* ScalarType.STRING */ },
+    { no: 5, name: "resources", kind: "message", T: Resources },
+  ]);
+
+  static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): AutoscalerOptions {
+    return new AutoscalerOptions().fromBinary(bytes, options);
+  }
+
+  static fromJson(jsonValue: JsonValue, options?: Partial<JsonReadOptions>): AutoscalerOptions {
+    return new AutoscalerOptions().fromJson(jsonValue, options);
+  }
+
+  static fromJsonString(jsonString: string, options?: Partial<JsonReadOptions>): AutoscalerOptions {
+    return new AutoscalerOptions().fromJsonString(jsonString, options);
+  }
+
+  static equals(a: AutoscalerOptions | PlainMessage<AutoscalerOptions> | undefined, b: AutoscalerOptions | PlainMessage<AutoscalerOptions> | undefined): boolean {
+    return proto3.util.equals(AutoscalerOptions, a, b);
+  }
+}
+
+/**
+ * @generated from enum flyteidl.plugins.AutoscalerOptions.UpscalingMode
+ */
+export enum AutoscalerOptions_UpscalingMode {
+  /**
+   * @generated from enum value: UPSCALING_MODE_UNSPECIFIED = 0;
+   */
+  UNSPECIFIED = 0,
+
+  /**
+   * @generated from enum value: UPSCALING_MODE_DEFAULT = 1;
+   */
+  DEFAULT = 1,
+
+  /**
+   * @generated from enum value: UPSCALING_MODE_AGGRESSIVE = 2;
+   */
+  AGGRESSIVE = 2,
+
+  /**
+   * @generated from enum value: UPSCALING_MODE_CONSERVATIVE = 3;
+   */
+  CONSERVATIVE = 3,
+}
+// Retrieve enum metadata with: proto3.getEnumType(AutoscalerOptions_UpscalingMode)
+proto3.util.setEnumType(AutoscalerOptions_UpscalingMode, "flyteidl.plugins.AutoscalerOptions.UpscalingMode", [
+  { no: 0, name: "UPSCALING_MODE_UNSPECIFIED" },
+  { no: 1, name: "UPSCALING_MODE_DEFAULT" },
+  { no: 2, name: "UPSCALING_MODE_AGGRESSIVE" },
+  { no: 3, name: "UPSCALING_MODE_CONSERVATIVE" },
+]);
+
 /**
  * Define Ray cluster defines the desired state of RayCluster
  *
@@ -110,6 +210,11 @@ export class RayCluster extends Message<RayCluster> {
    */
   enableAutoscaling = false;
 
+  /**
+   * @generated from field: flyteidl.plugins.AutoscalerOptions autoscaler_options = 4;
+   */
+  autoscalerOptions?: AutoscalerOptions;
+
   constructor(data?: PartialMessage<RayCluster>) {
     super();
     proto3.util.initPartial(data, this);
@@ -121,6 +226,7 @@ export class RayCluster extends Message<RayCluster> {
     { no: 1, name: "head_group_spec", kind: "message", T: HeadGroupSpec },
     { no: 2, name: "worker_group_spec", kind: "message", T: WorkerGroupSpec, repeated: true },
     { no: 3, name: "enable_autoscaling", kind: "scalar", T: 8 /* ScalarType.BOOL */ },
+    { no: 4, name: "autoscaler_options", kind: "message", T: AutoscalerOptions },
   ]);
 
   static fromBinary(bytes: Uint8Array, options?: Partial<BinaryReadOptions>): RayCluster {
diff --git a/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.go b/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.go
index 18d0c4c1cba..26f946a1efe 100644
--- a/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.go
+++ b/flyteidl/gen/pb-go/flyteidl/plugins/ray.pb.go
@@ -21,6 +21,58 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
+type AutoscalerOptions_UpscalingMode int32
+
+const (
+	AutoscalerOptions_UPSCALING_MODE_UNSPECIFIED  AutoscalerOptions_UpscalingMode = 0
+	AutoscalerOptions_UPSCALING_MODE_DEFAULT      AutoscalerOptions_UpscalingMode = 1
+	AutoscalerOptions_UPSCALING_MODE_AGGRESSIVE   AutoscalerOptions_UpscalingMode = 2
+	AutoscalerOptions_UPSCALING_MODE_CONSERVATIVE AutoscalerOptions_UpscalingMode = 3
+)
+
+// Enum value maps for AutoscalerOptions_UpscalingMode.
+var (
+	AutoscalerOptions_UpscalingMode_name = map[int32]string{
+		0: "UPSCALING_MODE_UNSPECIFIED",
+		1: "UPSCALING_MODE_DEFAULT",
+		2: "UPSCALING_MODE_AGGRESSIVE",
+		3: "UPSCALING_MODE_CONSERVATIVE",
+	}
+	AutoscalerOptions_UpscalingMode_value = map[string]int32{
+		"UPSCALING_MODE_UNSPECIFIED":  0,
+		"UPSCALING_MODE_DEFAULT":      1,
+		"UPSCALING_MODE_AGGRESSIVE":   2,
+		"UPSCALING_MODE_CONSERVATIVE": 3,
+	}
+)
+
+func (x AutoscalerOptions_UpscalingMode) Enum() *AutoscalerOptions_UpscalingMode {
+	p := new(AutoscalerOptions_UpscalingMode)
+	*p = x
+	return p
+}
+
+func (x AutoscalerOptions_UpscalingMode) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (AutoscalerOptions_UpscalingMode) Descriptor() protoreflect.EnumDescriptor {
+	return file_flyteidl_plugins_ray_proto_enumTypes[0].Descriptor()
+}
+
+func (AutoscalerOptions_UpscalingMode) Type() protoreflect.EnumType {
+	return &file_flyteidl_plugins_ray_proto_enumTypes[0]
+}
+
+func (x AutoscalerOptions_UpscalingMode) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use AutoscalerOptions_UpscalingMode.Descriptor instead.
+func (AutoscalerOptions_UpscalingMode) EnumDescriptor() ([]byte, []int) {
+	return file_flyteidl_plugins_ray_proto_rawDescGZIP(), []int{1, 0}
+}
+
 // RayJobSpec defines the desired state of RayJob
 type RayJob struct {
 	state         protoimpl.MessageState
@@ -111,6 +163,88 @@ func (x *RayJob) GetRuntimeEnvYaml() string {
 	return ""
 }
 
+type AutoscalerOptions struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	UpscalingMode      AutoscalerOptions_UpscalingMode `protobuf:"varint,1,opt,name=upscaling_mode,json=upscalingMode,proto3,enum=flyteidl.plugins.AutoscalerOptions_UpscalingMode" json:"upscaling_mode,omitempty"`
+	IdleTimeoutSeconds int32                           `protobuf:"varint,2,opt,name=idle_timeout_seconds,json=idleTimeoutSeconds,proto3" json:"idle_timeout_seconds,omitempty"`
+	// autoscaler sidecar env vars
+	Env []*core.KeyValuePair `protobuf:"bytes,3,rep,name=env,proto3" json:"env,omitempty"`
+	// custom autoscaler image
+	Image string `protobuf:"bytes,4,opt,name=image,proto3" json:"image,omitempty"`
+	// autoscaler container resources
+	Resources *core.Resources `protobuf:"bytes,5,opt,name=resources,proto3" json:"resources,omitempty"`
+}
+
+func (x *AutoscalerOptions) Reset() {
+	*x = AutoscalerOptions{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_flyteidl_plugins_ray_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *AutoscalerOptions) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*AutoscalerOptions) ProtoMessage() {}
+
+func (x *AutoscalerOptions) ProtoReflect() protoreflect.Message {
+	mi := &file_flyteidl_plugins_ray_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use AutoscalerOptions.ProtoReflect.Descriptor instead.
+func (*AutoscalerOptions) Descriptor() ([]byte, []int) {
+	return file_flyteidl_plugins_ray_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *AutoscalerOptions) GetUpscalingMode() AutoscalerOptions_UpscalingMode {
+	if x != nil {
+		return x.UpscalingMode
+	}
+	return AutoscalerOptions_UPSCALING_MODE_UNSPECIFIED
+}
+
+func (x *AutoscalerOptions) GetIdleTimeoutSeconds() int32 {
+	if x != nil {
+		return x.IdleTimeoutSeconds
+	}
+	return 0
+}
+
+func (x *AutoscalerOptions) GetEnv() []*core.KeyValuePair {
+	if x != nil {
+		return x.Env
+	}
+	return nil
+}
+
+func (x *AutoscalerOptions) GetImage() string {
+	if x != nil {
+		return x.Image
+	}
+	return ""
+}
+
+func (x *AutoscalerOptions) GetResources() *core.Resources {
+	if x != nil {
+		return x.Resources
+	}
+	return nil
+}
+
 // Define Ray cluster defines the desired state of RayCluster
 type RayCluster struct {
 	state         protoimpl.MessageState
@@ -122,13 +256,14 @@ type RayCluster struct {
 	// WorkerGroupSpecs are the specs for the worker pods
 	WorkerGroupSpec []*WorkerGroupSpec `protobuf:"bytes,2,rep,name=worker_group_spec,json=workerGroupSpec,proto3" json:"worker_group_spec,omitempty"`
 	// Whether to enable autoscaling.
-	EnableAutoscaling bool `protobuf:"varint,3,opt,name=enable_autoscaling,json=enableAutoscaling,proto3" json:"enable_autoscaling,omitempty"`
+	EnableAutoscaling bool               `protobuf:"varint,3,opt,name=enable_autoscaling,json=enableAutoscaling,proto3" json:"enable_autoscaling,omitempty"`
+	AutoscalerOptions *AutoscalerOptions `protobuf:"bytes,4,opt,name=autoscaler_options,json=autoscalerOptions,proto3" json:"autoscaler_options,omitempty"`
 }
 
 func (x *RayCluster) Reset() {
 	*x = RayCluster{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_flyteidl_plugins_ray_proto_msgTypes[1]
+		mi := &file_flyteidl_plugins_ray_proto_msgTypes[2]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -141,7 +276,7 @@ func (x *RayCluster) String() string {
 func (*RayCluster) ProtoMessage() {}
 
 func (x *RayCluster) ProtoReflect() protoreflect.Message {
-	mi := &file_flyteidl_plugins_ray_proto_msgTypes[1]
+	mi := &file_flyteidl_plugins_ray_proto_msgTypes[2]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -154,7 +289,7 @@ func (x *RayCluster) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use RayCluster.ProtoReflect.Descriptor instead.
 func (*RayCluster) Descriptor() ([]byte, []int) {
-	return file_flyteidl_plugins_ray_proto_rawDescGZIP(), []int{1}
+	return file_flyteidl_plugins_ray_proto_rawDescGZIP(), []int{2}
 }
 
 func (x *RayCluster) GetHeadGroupSpec() *HeadGroupSpec {
@@ -178,6 +313,13 @@ func (x *RayCluster) GetEnableAutoscaling() bool {
 	return false
 }
 
+func (x *RayCluster) GetAutoscalerOptions() *AutoscalerOptions {
+	if x != nil {
+		return x.AutoscalerOptions
+	}
+	return nil
+}
+
 // HeadGroupSpec are the spec for the head pod
 type HeadGroupSpec struct {
 	state         protoimpl.MessageState
@@ -194,7 +336,7 @@ type HeadGroupSpec struct {
 func (x *HeadGroupSpec) Reset() {
 	*x = HeadGroupSpec{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_flyteidl_plugins_ray_proto_msgTypes[2]
+		mi := &file_flyteidl_plugins_ray_proto_msgTypes[3]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -207,7 +349,7 @@ func (x *HeadGroupSpec) String() string {
 func (*HeadGroupSpec) ProtoMessage() {}
 
 func (x *HeadGroupSpec) ProtoReflect() protoreflect.Message {
-	mi := &file_flyteidl_plugins_ray_proto_msgTypes[2]
+	mi := &file_flyteidl_plugins_ray_proto_msgTypes[3]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -220,7 +362,7 @@ func (x *HeadGroupSpec) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use HeadGroupSpec.ProtoReflect.Descriptor instead.
 func (*HeadGroupSpec) Descriptor() ([]byte, []int) {
-	return file_flyteidl_plugins_ray_proto_rawDescGZIP(), []int{2}
+	return file_flyteidl_plugins_ray_proto_rawDescGZIP(), []int{3}
 }
 
 func (x *HeadGroupSpec) GetRayStartParams() map[string]string {
@@ -261,7 +403,7 @@ type WorkerGroupSpec struct {
 func (x *WorkerGroupSpec) Reset() {
 	*x = WorkerGroupSpec{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_flyteidl_plugins_ray_proto_msgTypes[3]
+		mi := &file_flyteidl_plugins_ray_proto_msgTypes[4]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -274,7 +416,7 @@ func (x *WorkerGroupSpec) String() string {
 func (*WorkerGroupSpec) ProtoMessage() {}
 
 func (x *WorkerGroupSpec) ProtoReflect() protoreflect.Message {
-	mi := &file_flyteidl_plugins_ray_proto_msgTypes[3]
+	mi := &file_flyteidl_plugins_ray_proto_msgTypes[4]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -287,7 +429,7 @@ func (x *WorkerGroupSpec) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use WorkerGroupSpec.ProtoReflect.Descriptor instead.
 func (*WorkerGroupSpec) Descriptor() ([]byte, []int) {
-	return file_flyteidl_plugins_ray_proto_rawDescGZIP(), []int{3}
+	return file_flyteidl_plugins_ray_proto_rawDescGZIP(), []int{4}
 }
 
 func (x *WorkerGroupSpec) GetGroupName() string {
@@ -339,87 +481,121 @@ var file_flyteidl_plugins_ray_proto_rawDesc = []byte{
 	0x6e, 0x73, 0x2f, 0x72, 0x61, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x10, 0x66, 0x6c,
 	0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x1a, 0x19,
 	0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2f, 0x63, 0x6f, 0x72, 0x65, 0x2f, 0x74, 0x61,
-	0x73, 0x6b, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x92, 0x02, 0x0a, 0x06, 0x52, 0x61,
-	0x79, 0x4a, 0x6f, 0x62, 0x12, 0x3d, 0x0a, 0x0b, 0x72, 0x61, 0x79, 0x5f, 0x63, 0x6c, 0x75, 0x73,
-	0x74, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x66, 0x6c, 0x79, 0x74,
-	0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x2e, 0x52, 0x61, 0x79,
-	0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x0a, 0x72, 0x61, 0x79, 0x43, 0x6c, 0x75, 0x73,
-	0x74, 0x65, 0x72, 0x12, 0x23, 0x0a, 0x0b, 0x72, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x5f, 0x65,
-	0x6e, 0x76, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x02, 0x18, 0x01, 0x52, 0x0a, 0x72, 0x75,
-	0x6e, 0x74, 0x69, 0x6d, 0x65, 0x45, 0x6e, 0x76, 0x12, 0x3d, 0x0a, 0x1b, 0x73, 0x68, 0x75, 0x74,
-	0x64, 0x6f, 0x77, 0x6e, 0x5f, 0x61, 0x66, 0x74, 0x65, 0x72, 0x5f, 0x6a, 0x6f, 0x62, 0x5f, 0x66,
-	0x69, 0x6e, 0x69, 0x73, 0x68, 0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x18, 0x73,
-	0x68, 0x75, 0x74, 0x64, 0x6f, 0x77, 0x6e, 0x41, 0x66, 0x74, 0x65, 0x72, 0x4a, 0x6f, 0x62, 0x46,
-	0x69, 0x6e, 0x69, 0x73, 0x68, 0x65, 0x73, 0x12, 0x3b, 0x0a, 0x1a, 0x74, 0x74, 0x6c, 0x5f, 0x73,
-	0x65, 0x63, 0x6f, 0x6e, 0x64, 0x73, 0x5f, 0x61, 0x66, 0x74, 0x65, 0x72, 0x5f, 0x66, 0x69, 0x6e,
-	0x69, 0x73, 0x68, 0x65, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x17, 0x74, 0x74, 0x6c,
-	0x53, 0x65, 0x63, 0x6f, 0x6e, 0x64, 0x73, 0x41, 0x66, 0x74, 0x65, 0x72, 0x46, 0x69, 0x6e, 0x69,
-	0x73, 0x68, 0x65, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x72, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x5f,
-	0x65, 0x6e, 0x76, 0x5f, 0x79, 0x61, 0x6d, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e,
-	0x72, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x45, 0x6e, 0x76, 0x59, 0x61, 0x6d, 0x6c, 0x22, 0xd3,
-	0x01, 0x0a, 0x0a, 0x52, 0x61, 0x79, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x47, 0x0a,
-	0x0f, 0x68, 0x65, 0x61, 0x64, 0x5f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x73, 0x70, 0x65, 0x63,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1f, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64,
-	0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x47, 0x72,
-	0x6f, 0x75, 0x70, 0x53, 0x70, 0x65, 0x63, 0x52, 0x0d, 0x68, 0x65, 0x61, 0x64, 0x47, 0x72, 0x6f,
-	0x75, 0x70, 0x53, 0x70, 0x65, 0x63, 0x12, 0x4d, 0x0a, 0x11, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72,
-	0x5f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x73, 0x70, 0x65, 0x63, 0x18, 0x02, 0x20, 0x03, 0x28,
-	0x0b, 0x32, 0x21, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75,
-	0x67, 0x69, 0x6e, 0x73, 0x2e, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70,
-	0x53, 0x70, 0x65, 0x63, 0x52, 0x0f, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75,
-	0x70, 0x53, 0x70, 0x65, 0x63, 0x12, 0x2d, 0x0a, 0x12, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x5f,
-	0x61, 0x75, 0x74, 0x6f, 0x73, 0x63, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0x18, 0x03, 0x20, 0x01, 0x28,
-	0x08, 0x52, 0x11, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x41, 0x75, 0x74, 0x6f, 0x73, 0x63, 0x61,
-	0x6c, 0x69, 0x6e, 0x67, 0x22, 0xe1, 0x01, 0x0a, 0x0d, 0x48, 0x65, 0x61, 0x64, 0x47, 0x72, 0x6f,
-	0x75, 0x70, 0x53, 0x70, 0x65, 0x63, 0x12, 0x5d, 0x0a, 0x10, 0x72, 0x61, 0x79, 0x5f, 0x73, 0x74,
-	0x61, 0x72, 0x74, 0x5f, 0x70, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b,
-	0x32, 0x33, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67,
-	0x69, 0x6e, 0x73, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65,
-	0x63, 0x2e, 0x52, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73,
-	0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0e, 0x72, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50,
-	0x61, 0x72, 0x61, 0x6d, 0x73, 0x12, 0x2e, 0x0a, 0x07, 0x6b, 0x38, 0x73, 0x5f, 0x70, 0x6f, 0x64,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64,
-	0x6c, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x4b, 0x38, 0x73, 0x50, 0x6f, 0x64, 0x52, 0x06, 0x6b,
-	0x38, 0x73, 0x50, 0x6f, 0x64, 0x1a, 0x41, 0x0a, 0x13, 0x52, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72,
-	0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03,
-	0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14,
-	0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76,
-	0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xe6, 0x02, 0x0a, 0x0f, 0x57, 0x6f, 0x72,
-	0x6b, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65, 0x63, 0x12, 0x1d, 0x0a, 0x0a,
-	0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x09, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x72,
-	0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x08, 0x72,
-	0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6d, 0x69, 0x6e, 0x5f, 0x72,
-	0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0b, 0x6d,
-	0x69, 0x6e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6d, 0x61,
-	0x78, 0x5f, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05,
-	0x52, 0x0b, 0x6d, 0x61, 0x78, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x12, 0x5f, 0x0a,
-	0x10, 0x72, 0x61, 0x79, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x5f, 0x70, 0x61, 0x72, 0x61, 0x6d,
-	0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x35, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69,
-	0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x2e, 0x57, 0x6f, 0x72, 0x6b, 0x65,
-	0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65, 0x63, 0x2e, 0x52, 0x61, 0x79, 0x53, 0x74,
-	0x61, 0x72, 0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0e,
-	0x72, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x12, 0x2e,
-	0x0a, 0x07, 0x6b, 0x38, 0x73, 0x5f, 0x70, 0x6f, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32,
-	0x15, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e,
-	0x4b, 0x38, 0x73, 0x50, 0x6f, 0x64, 0x52, 0x06, 0x6b, 0x38, 0x73, 0x50, 0x6f, 0x64, 0x1a, 0x41,
-	0x0a, 0x13, 0x52, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73,
-	0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38,
-	0x01, 0x42, 0xc0, 0x01, 0x0a, 0x14, 0x63, 0x6f, 0x6d, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69,
-	0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x42, 0x08, 0x52, 0x61, 0x79, 0x50,
-	0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x3d, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63,
-	0x6f, 0x6d, 0x2f, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x6f, 0x72, 0x67, 0x2f, 0x66, 0x6c, 0x79, 0x74,
-	0x65, 0x2f, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x70,
-	0x62, 0x2d, 0x67, 0x6f, 0x2f, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2f, 0x70, 0x6c,
-	0x75, 0x67, 0x69, 0x6e, 0x73, 0xa2, 0x02, 0x03, 0x46, 0x50, 0x58, 0xaa, 0x02, 0x10, 0x46, 0x6c,
-	0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0xca, 0x02,
-	0x10, 0x46, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x5c, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e,
-	0x73, 0xe2, 0x02, 0x1c, 0x46, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x5c, 0x50, 0x6c, 0x75,
-	0x67, 0x69, 0x6e, 0x73, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61,
-	0xea, 0x02, 0x11, 0x46, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x3a, 0x3a, 0x50, 0x6c, 0x75,
-	0x67, 0x69, 0x6e, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x73, 0x6b, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1c, 0x66, 0x6c, 0x79, 0x74, 0x65,
+	0x69, 0x64, 0x6c, 0x2f, 0x63, 0x6f, 0x72, 0x65, 0x2f, 0x6c, 0x69, 0x74, 0x65, 0x72, 0x61, 0x6c,
+	0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x92, 0x02, 0x0a, 0x06, 0x52, 0x61, 0x79, 0x4a,
+	0x6f, 0x62, 0x12, 0x3d, 0x0a, 0x0b, 0x72, 0x61, 0x79, 0x5f, 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65,
+	0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69,
+	0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x2e, 0x52, 0x61, 0x79, 0x43, 0x6c,
+	0x75, 0x73, 0x74, 0x65, 0x72, 0x52, 0x0a, 0x72, 0x61, 0x79, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65,
+	0x72, 0x12, 0x23, 0x0a, 0x0b, 0x72, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x5f, 0x65, 0x6e, 0x76,
+	0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x42, 0x02, 0x18, 0x01, 0x52, 0x0a, 0x72, 0x75, 0x6e, 0x74,
+	0x69, 0x6d, 0x65, 0x45, 0x6e, 0x76, 0x12, 0x3d, 0x0a, 0x1b, 0x73, 0x68, 0x75, 0x74, 0x64, 0x6f,
+	0x77, 0x6e, 0x5f, 0x61, 0x66, 0x74, 0x65, 0x72, 0x5f, 0x6a, 0x6f, 0x62, 0x5f, 0x66, 0x69, 0x6e,
+	0x69, 0x73, 0x68, 0x65, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x18, 0x73, 0x68, 0x75,
+	0x74, 0x64, 0x6f, 0x77, 0x6e, 0x41, 0x66, 0x74, 0x65, 0x72, 0x4a, 0x6f, 0x62, 0x46, 0x69, 0x6e,
+	0x69, 0x73, 0x68, 0x65, 0x73, 0x12, 0x3b, 0x0a, 0x1a, 0x74, 0x74, 0x6c, 0x5f, 0x73, 0x65, 0x63,
+	0x6f, 0x6e, 0x64, 0x73, 0x5f, 0x61, 0x66, 0x74, 0x65, 0x72, 0x5f, 0x66, 0x69, 0x6e, 0x69, 0x73,
+	0x68, 0x65, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x17, 0x74, 0x74, 0x6c, 0x53, 0x65,
+	0x63, 0x6f, 0x6e, 0x64, 0x73, 0x41, 0x66, 0x74, 0x65, 0x72, 0x46, 0x69, 0x6e, 0x69, 0x73, 0x68,
+	0x65, 0x64, 0x12, 0x28, 0x0a, 0x10, 0x72, 0x75, 0x6e, 0x74, 0x69, 0x6d, 0x65, 0x5f, 0x65, 0x6e,
+	0x76, 0x5f, 0x79, 0x61, 0x6d, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x72, 0x75,
+	0x6e, 0x74, 0x69, 0x6d, 0x65, 0x45, 0x6e, 0x76, 0x59, 0x61, 0x6d, 0x6c, 0x22, 0xaa, 0x03, 0x0a,
+	0x11, 0x41, 0x75, 0x74, 0x6f, 0x73, 0x63, 0x61, 0x6c, 0x65, 0x72, 0x4f, 0x70, 0x74, 0x69, 0x6f,
+	0x6e, 0x73, 0x12, 0x58, 0x0a, 0x0e, 0x75, 0x70, 0x73, 0x63, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0x5f,
+	0x6d, 0x6f, 0x64, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x31, 0x2e, 0x66, 0x6c, 0x79,
+	0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x2e, 0x41, 0x75,
+	0x74, 0x6f, 0x73, 0x63, 0x61, 0x6c, 0x65, 0x72, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x2e,
+	0x55, 0x70, 0x73, 0x63, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0x4d, 0x6f, 0x64, 0x65, 0x52, 0x0d, 0x75,
+	0x70, 0x73, 0x63, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x30, 0x0a, 0x14,
+	0x69, 0x64, 0x6c, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x5f, 0x73, 0x65, 0x63,
+	0x6f, 0x6e, 0x64, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x12, 0x69, 0x64, 0x6c, 0x65,
+	0x54, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x53, 0x65, 0x63, 0x6f, 0x6e, 0x64, 0x73, 0x12, 0x2d,
+	0x0a, 0x03, 0x65, 0x6e, 0x76, 0x18, 0x03, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1b, 0x2e, 0x66, 0x6c,
+	0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x4b, 0x65, 0x79, 0x56,
+	0x61, 0x6c, 0x75, 0x65, 0x50, 0x61, 0x69, 0x72, 0x52, 0x03, 0x65, 0x6e, 0x76, 0x12, 0x14, 0x0a,
+	0x05, 0x69, 0x6d, 0x61, 0x67, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x69, 0x6d,
+	0x61, 0x67, 0x65, 0x12, 0x36, 0x0a, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73,
+	0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x18, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64,
+	0x6c, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73,
+	0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x22, 0x8b, 0x01, 0x0a, 0x0d,
+	0x55, 0x70, 0x73, 0x63, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x1e, 0x0a,
+	0x1a, 0x55, 0x50, 0x53, 0x43, 0x41, 0x4c, 0x49, 0x4e, 0x47, 0x5f, 0x4d, 0x4f, 0x44, 0x45, 0x5f,
+	0x55, 0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x1a, 0x0a,
+	0x16, 0x55, 0x50, 0x53, 0x43, 0x41, 0x4c, 0x49, 0x4e, 0x47, 0x5f, 0x4d, 0x4f, 0x44, 0x45, 0x5f,
+	0x44, 0x45, 0x46, 0x41, 0x55, 0x4c, 0x54, 0x10, 0x01, 0x12, 0x1d, 0x0a, 0x19, 0x55, 0x50, 0x53,
+	0x43, 0x41, 0x4c, 0x49, 0x4e, 0x47, 0x5f, 0x4d, 0x4f, 0x44, 0x45, 0x5f, 0x41, 0x47, 0x47, 0x52,
+	0x45, 0x53, 0x53, 0x49, 0x56, 0x45, 0x10, 0x02, 0x12, 0x1f, 0x0a, 0x1b, 0x55, 0x50, 0x53, 0x43,
+	0x41, 0x4c, 0x49, 0x4e, 0x47, 0x5f, 0x4d, 0x4f, 0x44, 0x45, 0x5f, 0x43, 0x4f, 0x4e, 0x53, 0x45,
+	0x52, 0x56, 0x41, 0x54, 0x49, 0x56, 0x45, 0x10, 0x03, 0x22, 0xa7, 0x02, 0x0a, 0x0a, 0x52, 0x61,
+	0x79, 0x43, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x12, 0x47, 0x0a, 0x0f, 0x68, 0x65, 0x61, 0x64,
+	0x5f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x73, 0x70, 0x65, 0x63, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x0b, 0x32, 0x1f, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75,
+	0x67, 0x69, 0x6e, 0x73, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70,
+	0x65, 0x63, 0x52, 0x0d, 0x68, 0x65, 0x61, 0x64, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65,
+	0x63, 0x12, 0x4d, 0x0a, 0x11, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x5f, 0x67, 0x72, 0x6f, 0x75,
+	0x70, 0x5f, 0x73, 0x70, 0x65, 0x63, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x21, 0x2e, 0x66,
+	0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x2e,
+	0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65, 0x63, 0x52,
+	0x0f, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65, 0x63,
+	0x12, 0x2d, 0x0a, 0x12, 0x65, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x5f, 0x61, 0x75, 0x74, 0x6f, 0x73,
+	0x63, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x11, 0x65, 0x6e,
+	0x61, 0x62, 0x6c, 0x65, 0x41, 0x75, 0x74, 0x6f, 0x73, 0x63, 0x61, 0x6c, 0x69, 0x6e, 0x67, 0x12,
+	0x52, 0x0a, 0x12, 0x61, 0x75, 0x74, 0x6f, 0x73, 0x63, 0x61, 0x6c, 0x65, 0x72, 0x5f, 0x6f, 0x70,
+	0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x23, 0x2e, 0x66, 0x6c,
+	0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x2e, 0x41,
+	0x75, 0x74, 0x6f, 0x73, 0x63, 0x61, 0x6c, 0x65, 0x72, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73,
+	0x52, 0x11, 0x61, 0x75, 0x74, 0x6f, 0x73, 0x63, 0x61, 0x6c, 0x65, 0x72, 0x4f, 0x70, 0x74, 0x69,
+	0x6f, 0x6e, 0x73, 0x22, 0xe1, 0x01, 0x0a, 0x0d, 0x48, 0x65, 0x61, 0x64, 0x47, 0x72, 0x6f, 0x75,
+	0x70, 0x53, 0x70, 0x65, 0x63, 0x12, 0x5d, 0x0a, 0x10, 0x72, 0x61, 0x79, 0x5f, 0x73, 0x74, 0x61,
+	0x72, 0x74, 0x5f, 0x70, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32,
+	0x33, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69,
+	0x6e, 0x73, 0x2e, 0x48, 0x65, 0x61, 0x64, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65, 0x63,
+	0x2e, 0x52, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x45,
+	0x6e, 0x74, 0x72, 0x79, 0x52, 0x0e, 0x72, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50, 0x61,
+	0x72, 0x61, 0x6d, 0x73, 0x12, 0x2e, 0x0a, 0x07, 0x6b, 0x38, 0x73, 0x5f, 0x70, 0x6f, 0x64, 0x18,
+	0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c,
+	0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x4b, 0x38, 0x73, 0x50, 0x6f, 0x64, 0x52, 0x06, 0x6b, 0x38,
+	0x73, 0x50, 0x6f, 0x64, 0x1a, 0x41, 0x0a, 0x13, 0x52, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74,
+	0x50, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b,
+	0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a,
+	0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61,
+	0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xe6, 0x02, 0x0a, 0x0f, 0x57, 0x6f, 0x72, 0x6b,
+	0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65, 0x63, 0x12, 0x1d, 0x0a, 0x0a, 0x67,
+	0x72, 0x6f, 0x75, 0x70, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x09, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x72, 0x65,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x08, 0x72, 0x65,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6d, 0x69, 0x6e, 0x5f, 0x72, 0x65,
+	0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0b, 0x6d, 0x69,
+	0x6e, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x12, 0x21, 0x0a, 0x0c, 0x6d, 0x61, 0x78,
+	0x5f, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52,
+	0x0b, 0x6d, 0x61, 0x78, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x73, 0x12, 0x5f, 0x0a, 0x10,
+	0x72, 0x61, 0x79, 0x5f, 0x73, 0x74, 0x61, 0x72, 0x74, 0x5f, 0x70, 0x61, 0x72, 0x61, 0x6d, 0x73,
+	0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x35, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64,
+	0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x2e, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72,
+	0x47, 0x72, 0x6f, 0x75, 0x70, 0x53, 0x70, 0x65, 0x63, 0x2e, 0x52, 0x61, 0x79, 0x53, 0x74, 0x61,
+	0x72, 0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0e, 0x72,
+	0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x12, 0x2e, 0x0a,
+	0x07, 0x6b, 0x38, 0x73, 0x5f, 0x70, 0x6f, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15,
+	0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x4b,
+	0x38, 0x73, 0x50, 0x6f, 0x64, 0x52, 0x06, 0x6b, 0x38, 0x73, 0x50, 0x6f, 0x64, 0x1a, 0x41, 0x0a,
+	0x13, 0x52, 0x61, 0x79, 0x53, 0x74, 0x61, 0x72, 0x74, 0x50, 0x61, 0x72, 0x61, 0x6d, 0x73, 0x45,
+	0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18,
+	0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01,
+	0x42, 0xc0, 0x01, 0x0a, 0x14, 0x63, 0x6f, 0x6d, 0x2e, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64,
+	0x6c, 0x2e, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0x42, 0x08, 0x52, 0x61, 0x79, 0x50, 0x72,
+	0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x3d, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f,
+	0x6d, 0x2f, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x6f, 0x72, 0x67, 0x2f, 0x66, 0x6c, 0x79, 0x74, 0x65,
+	0x2f, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2f, 0x67, 0x65, 0x6e, 0x2f, 0x70, 0x62,
+	0x2d, 0x67, 0x6f, 0x2f, 0x66, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x2f, 0x70, 0x6c, 0x75,
+	0x67, 0x69, 0x6e, 0x73, 0xa2, 0x02, 0x03, 0x46, 0x50, 0x58, 0xaa, 0x02, 0x10, 0x46, 0x6c, 0x79,
+	0x74, 0x65, 0x69, 0x64, 0x6c, 0x2e, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73, 0xca, 0x02, 0x10,
+	0x46, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x5c, 0x50, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x73,
+	0xe2, 0x02, 0x1c, 0x46, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x5c, 0x50, 0x6c, 0x75, 0x67,
+	0x69, 0x6e, 0x73, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea,
+	0x02, 0x11, 0x46, 0x6c, 0x79, 0x74, 0x65, 0x69, 0x64, 0x6c, 0x3a, 0x3a, 0x50, 0x6c, 0x75, 0x67,
+	0x69, 0x6e, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -434,29 +610,38 @@ func file_flyteidl_plugins_ray_proto_rawDescGZIP() []byte {
 	return file_flyteidl_plugins_ray_proto_rawDescData
 }
 
-var file_flyteidl_plugins_ray_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
+var file_flyteidl_plugins_ray_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_flyteidl_plugins_ray_proto_msgTypes = make([]protoimpl.MessageInfo, 7)
 var file_flyteidl_plugins_ray_proto_goTypes = []interface{}{
-	(*RayJob)(nil),          // 0: flyteidl.plugins.RayJob
-	(*RayCluster)(nil),      // 1: flyteidl.plugins.RayCluster
-	(*HeadGroupSpec)(nil),   // 2: flyteidl.plugins.HeadGroupSpec
-	(*WorkerGroupSpec)(nil), // 3: flyteidl.plugins.WorkerGroupSpec
-	nil,                     // 4: flyteidl.plugins.HeadGroupSpec.RayStartParamsEntry
-	nil,                     // 5: flyteidl.plugins.WorkerGroupSpec.RayStartParamsEntry
-	(*core.K8SPod)(nil),     // 6: flyteidl.core.K8sPod
+	(AutoscalerOptions_UpscalingMode)(0), // 0: flyteidl.plugins.AutoscalerOptions.UpscalingMode
+	(*RayJob)(nil),                       // 1: flyteidl.plugins.RayJob
+	(*AutoscalerOptions)(nil),            // 2: flyteidl.plugins.AutoscalerOptions
+	(*RayCluster)(nil),                   // 3: flyteidl.plugins.RayCluster
+	(*HeadGroupSpec)(nil),                // 4: flyteidl.plugins.HeadGroupSpec
+	(*WorkerGroupSpec)(nil),              // 5: flyteidl.plugins.WorkerGroupSpec
+	nil,                                  // 6: flyteidl.plugins.HeadGroupSpec.RayStartParamsEntry
+	nil,                                  // 7: flyteidl.plugins.WorkerGroupSpec.RayStartParamsEntry
+	(*core.KeyValuePair)(nil),            // 8: flyteidl.core.KeyValuePair
+	(*core.Resources)(nil),               // 9: flyteidl.core.Resources
+	(*core.K8SPod)(nil),                  // 10: flyteidl.core.K8sPod
 }
 var file_flyteidl_plugins_ray_proto_depIdxs = []int32{
-	1, // 0: flyteidl.plugins.RayJob.ray_cluster:type_name -> flyteidl.plugins.RayCluster
-	2, // 1: flyteidl.plugins.RayCluster.head_group_spec:type_name -> flyteidl.plugins.HeadGroupSpec
-	3, // 2: flyteidl.plugins.RayCluster.worker_group_spec:type_name -> flyteidl.plugins.WorkerGroupSpec
-	4, // 3: flyteidl.plugins.HeadGroupSpec.ray_start_params:type_name -> flyteidl.plugins.HeadGroupSpec.RayStartParamsEntry
-	6, // 4: flyteidl.plugins.HeadGroupSpec.k8s_pod:type_name -> flyteidl.core.K8sPod
-	5, // 5: flyteidl.plugins.WorkerGroupSpec.ray_start_params:type_name -> flyteidl.plugins.WorkerGroupSpec.RayStartParamsEntry
-	6, // 6: flyteidl.plugins.WorkerGroupSpec.k8s_pod:type_name -> flyteidl.core.K8sPod
-	7, // [7:7] is the sub-list for method output_type
-	7, // [7:7] is the sub-list for method input_type
-	7, // [7:7] is the sub-list for extension type_name
-	7, // [7:7] is the sub-list for extension extendee
-	0, // [0:7] is the sub-list for field type_name
+	3,  // 0: flyteidl.plugins.RayJob.ray_cluster:type_name -> flyteidl.plugins.RayCluster
+	0,  // 1: flyteidl.plugins.AutoscalerOptions.upscaling_mode:type_name -> flyteidl.plugins.AutoscalerOptions.UpscalingMode
+	8,  // 2: flyteidl.plugins.AutoscalerOptions.env:type_name -> flyteidl.core.KeyValuePair
+	9,  // 3: flyteidl.plugins.AutoscalerOptions.resources:type_name -> flyteidl.core.Resources
+	4,  // 4: flyteidl.plugins.RayCluster.head_group_spec:type_name -> flyteidl.plugins.HeadGroupSpec
+	5,  // 5: flyteidl.plugins.RayCluster.worker_group_spec:type_name -> flyteidl.plugins.WorkerGroupSpec
+	2,  // 6: flyteidl.plugins.RayCluster.autoscaler_options:type_name -> flyteidl.plugins.AutoscalerOptions
+	6,  // 7: flyteidl.plugins.HeadGroupSpec.ray_start_params:type_name -> flyteidl.plugins.HeadGroupSpec.RayStartParamsEntry
+	10, // 8: flyteidl.plugins.HeadGroupSpec.k8s_pod:type_name -> flyteidl.core.K8sPod
+	7,  // 9: flyteidl.plugins.WorkerGroupSpec.ray_start_params:type_name -> flyteidl.plugins.WorkerGroupSpec.RayStartParamsEntry
+	10, // 10: flyteidl.plugins.WorkerGroupSpec.k8s_pod:type_name -> flyteidl.core.K8sPod
+	11, // [11:11] is the sub-list for method output_type
+	11, // [11:11] is the sub-list for method input_type
+	11, // [11:11] is the sub-list for extension type_name
+	11, // [11:11] is the sub-list for extension extendee
+	0,  // [0:11] is the sub-list for field type_name
 }
 
 func init() { file_flyteidl_plugins_ray_proto_init() }
@@ -478,7 +663,7 @@ func file_flyteidl_plugins_ray_proto_init() {
 			}
 		}
 		file_flyteidl_plugins_ray_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*RayCluster); i {
+			switch v := v.(*AutoscalerOptions); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -490,7 +675,7 @@ func file_flyteidl_plugins_ray_proto_init() {
 			}
 		}
 		file_flyteidl_plugins_ray_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*HeadGroupSpec); i {
+			switch v := v.(*RayCluster); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -502,6 +687,18 @@ func file_flyteidl_plugins_ray_proto_init() {
 			}
 		}
 		file_flyteidl_plugins_ray_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*HeadGroupSpec); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_flyteidl_plugins_ray_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*WorkerGroupSpec); i {
 			case 0:
 				return &v.state
@@ -519,13 +716,14 @@ func file_flyteidl_plugins_ray_proto_init() {
 		File: protoimpl.DescBuilder{
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_flyteidl_plugins_ray_proto_rawDesc,
-			NumEnums:      0,
-			NumMessages:   6,
+			NumEnums:      1,
+			NumMessages:   7,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
 		GoTypes:           file_flyteidl_plugins_ray_proto_goTypes,
 		DependencyIndexes: file_flyteidl_plugins_ray_proto_depIdxs,
+		EnumInfos:         file_flyteidl_plugins_ray_proto_enumTypes,
 		MessageInfos:      file_flyteidl_plugins_ray_proto_msgTypes,
 	}.Build()
 	File_flyteidl_plugins_ray_proto = out.File
diff --git a/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.py b/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.py
index c625fd957b6..b1df12499f5 100644
--- a/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.py
+++ b/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.py
@@ -12,9 +12,10 @@
 
 
 from flyteidl.core import tasks_pb2 as flyteidl_dot_core_dot_tasks__pb2
+from flyteidl.core import literals_pb2 as flyteidl_dot_core_dot_literals__pb2
 
 
-DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1a\x66lyteidl/plugins/ray.proto\x12\x10\x66lyteidl.plugins\x1a\x19\x66lyteidl/core/tasks.proto\"\x92\x02\n\x06RayJob\x12=\n\x0bray_cluster\x18\x01 \x01(\x0b\x32\x1c.flyteidl.plugins.RayClusterR\nrayCluster\x12#\n\x0bruntime_env\x18\x02 \x01(\tB\x02\x18\x01R\nruntimeEnv\x12=\n\x1bshutdown_after_job_finishes\x18\x03 \x01(\x08R\x18shutdownAfterJobFinishes\x12;\n\x1attl_seconds_after_finished\x18\x04 \x01(\x05R\x17ttlSecondsAfterFinished\x12(\n\x10runtime_env_yaml\x18\x05 \x01(\tR\x0eruntimeEnvYaml\"\xd3\x01\n\nRayCluster\x12G\n\x0fhead_group_spec\x18\x01 \x01(\x0b\x32\x1f.flyteidl.plugins.HeadGroupSpecR\rheadGroupSpec\x12M\n\x11worker_group_spec\x18\x02 \x03(\x0b\x32!.flyteidl.plugins.WorkerGroupSpecR\x0fworkerGroupSpec\x12-\n\x12\x65nable_autoscaling\x18\x03 \x01(\x08R\x11\x65nableAutoscaling\"\xe1\x01\n\rHeadGroupSpec\x12]\n\x10ray_start_params\x18\x01 \x03(\x0b\x32\x33.flyteidl.plugins.HeadGroupSpec.RayStartParamsEntryR\x0erayStartParams\x12.\n\x07k8s_pod\x18\x02 \x01(\x0b\x32\x15.flyteidl.core.K8sPodR\x06k8sPod\x1a\x41\n\x13RayStartParamsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\xe6\x02\n\x0fWorkerGroupSpec\x12\x1d\n\ngroup_name\x18\x01 \x01(\tR\tgroupName\x12\x1a\n\x08replicas\x18\x02 \x01(\x05R\x08replicas\x12!\n\x0cmin_replicas\x18\x03 \x01(\x05R\x0bminReplicas\x12!\n\x0cmax_replicas\x18\x04 \x01(\x05R\x0bmaxReplicas\x12_\n\x10ray_start_params\x18\x05 \x03(\x0b\x32\x35.flyteidl.plugins.WorkerGroupSpec.RayStartParamsEntryR\x0erayStartParams\x12.\n\x07k8s_pod\x18\x06 \x01(\x0b\x32\x15.flyteidl.core.K8sPodR\x06k8sPod\x1a\x41\n\x13RayStartParamsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\xc0\x01\n\x14\x63om.flyteidl.pluginsB\x08RayProtoP\x01Z=github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/plugins\xa2\x02\x03\x46PX\xaa\x02\x10\x46lyteidl.Plugins\xca\x02\x10\x46lyteidl\\Plugins\xe2\x02\x1c\x46lyteidl\\Plugins\\GPBMetadata\xea\x02\x11\x46lyteidl::Pluginsb\x06proto3')
+DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1a\x66lyteidl/plugins/ray.proto\x12\x10\x66lyteidl.plugins\x1a\x19\x66lyteidl/core/tasks.proto\x1a\x1c\x66lyteidl/core/literals.proto\"\x92\x02\n\x06RayJob\x12=\n\x0bray_cluster\x18\x01 \x01(\x0b\x32\x1c.flyteidl.plugins.RayClusterR\nrayCluster\x12#\n\x0bruntime_env\x18\x02 \x01(\tB\x02\x18\x01R\nruntimeEnv\x12=\n\x1bshutdown_after_job_finishes\x18\x03 \x01(\x08R\x18shutdownAfterJobFinishes\x12;\n\x1attl_seconds_after_finished\x18\x04 \x01(\x05R\x17ttlSecondsAfterFinished\x12(\n\x10runtime_env_yaml\x18\x05 \x01(\tR\x0eruntimeEnvYaml\"\xaa\x03\n\x11\x41utoscalerOptions\x12X\n\x0eupscaling_mode\x18\x01 \x01(\x0e\x32\x31.flyteidl.plugins.AutoscalerOptions.UpscalingModeR\rupscalingMode\x12\x30\n\x14idle_timeout_seconds\x18\x02 \x01(\x05R\x12idleTimeoutSeconds\x12-\n\x03\x65nv\x18\x03 \x03(\x0b\x32\x1b.flyteidl.core.KeyValuePairR\x03\x65nv\x12\x14\n\x05image\x18\x04 \x01(\tR\x05image\x12\x36\n\tresources\x18\x05 \x01(\x0b\x32\x18.flyteidl.core.ResourcesR\tresources\"\x8b\x01\n\rUpscalingMode\x12\x1e\n\x1aUPSCALING_MODE_UNSPECIFIED\x10\x00\x12\x1a\n\x16UPSCALING_MODE_DEFAULT\x10\x01\x12\x1d\n\x19UPSCALING_MODE_AGGRESSIVE\x10\x02\x12\x1f\n\x1bUPSCALING_MODE_CONSERVATIVE\x10\x03\"\xa7\x02\n\nRayCluster\x12G\n\x0fhead_group_spec\x18\x01 \x01(\x0b\x32\x1f.flyteidl.plugins.HeadGroupSpecR\rheadGroupSpec\x12M\n\x11worker_group_spec\x18\x02 \x03(\x0b\x32!.flyteidl.plugins.WorkerGroupSpecR\x0fworkerGroupSpec\x12-\n\x12\x65nable_autoscaling\x18\x03 \x01(\x08R\x11\x65nableAutoscaling\x12R\n\x12\x61utoscaler_options\x18\x04 \x01(\x0b\x32#.flyteidl.plugins.AutoscalerOptionsR\x11\x61utoscalerOptions\"\xe1\x01\n\rHeadGroupSpec\x12]\n\x10ray_start_params\x18\x01 \x03(\x0b\x32\x33.flyteidl.plugins.HeadGroupSpec.RayStartParamsEntryR\x0erayStartParams\x12.\n\x07k8s_pod\x18\x02 \x01(\x0b\x32\x15.flyteidl.core.K8sPodR\x06k8sPod\x1a\x41\n\x13RayStartParamsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\"\xe6\x02\n\x0fWorkerGroupSpec\x12\x1d\n\ngroup_name\x18\x01 \x01(\tR\tgroupName\x12\x1a\n\x08replicas\x18\x02 \x01(\x05R\x08replicas\x12!\n\x0cmin_replicas\x18\x03 \x01(\x05R\x0bminReplicas\x12!\n\x0cmax_replicas\x18\x04 \x01(\x05R\x0bmaxReplicas\x12_\n\x10ray_start_params\x18\x05 \x03(\x0b\x32\x35.flyteidl.plugins.WorkerGroupSpec.RayStartParamsEntryR\x0erayStartParams\x12.\n\x07k8s_pod\x18\x06 \x01(\x0b\x32\x15.flyteidl.core.K8sPodR\x06k8sPod\x1a\x41\n\x13RayStartParamsEntry\x12\x10\n\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n\x05value\x18\x02 \x01(\tR\x05value:\x02\x38\x01\x42\xc0\x01\n\x14\x63om.flyteidl.pluginsB\x08RayProtoP\x01Z=github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/plugins\xa2\x02\x03\x46PX\xaa\x02\x10\x46lyteidl.Plugins\xca\x02\x10\x46lyteidl\\Plugins\xe2\x02\x1c\x46lyteidl\\Plugins\\GPBMetadata\xea\x02\x11\x46lyteidl::Pluginsb\x06proto3')
 
 _globals = globals()
 _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
@@ -29,16 +30,20 @@
   _HEADGROUPSPEC_RAYSTARTPARAMSENTRY._serialized_options = b'8\001'
   _WORKERGROUPSPEC_RAYSTARTPARAMSENTRY._options = None
   _WORKERGROUPSPEC_RAYSTARTPARAMSENTRY._serialized_options = b'8\001'
-  _globals['_RAYJOB']._serialized_start=76
-  _globals['_RAYJOB']._serialized_end=350
-  _globals['_RAYCLUSTER']._serialized_start=353
-  _globals['_RAYCLUSTER']._serialized_end=564
-  _globals['_HEADGROUPSPEC']._serialized_start=567
-  _globals['_HEADGROUPSPEC']._serialized_end=792
-  _globals['_HEADGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_start=727
-  _globals['_HEADGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_end=792
-  _globals['_WORKERGROUPSPEC']._serialized_start=795
-  _globals['_WORKERGROUPSPEC']._serialized_end=1153
-  _globals['_WORKERGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_start=727
-  _globals['_WORKERGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_end=792
+  _globals['_RAYJOB']._serialized_start=106
+  _globals['_RAYJOB']._serialized_end=380
+  _globals['_AUTOSCALEROPTIONS']._serialized_start=383
+  _globals['_AUTOSCALEROPTIONS']._serialized_end=809
+  _globals['_AUTOSCALEROPTIONS_UPSCALINGMODE']._serialized_start=670
+  _globals['_AUTOSCALEROPTIONS_UPSCALINGMODE']._serialized_end=809
+  _globals['_RAYCLUSTER']._serialized_start=812
+  _globals['_RAYCLUSTER']._serialized_end=1107
+  _globals['_HEADGROUPSPEC']._serialized_start=1110
+  _globals['_HEADGROUPSPEC']._serialized_end=1335
+  _globals['_HEADGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_start=1270
+  _globals['_HEADGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_end=1335
+  _globals['_WORKERGROUPSPEC']._serialized_start=1338
+  _globals['_WORKERGROUPSPEC']._serialized_end=1696
+  _globals['_WORKERGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_start=1270
+  _globals['_WORKERGROUPSPEC_RAYSTARTPARAMSENTRY']._serialized_end=1335
 # @@protoc_insertion_point(module_scope)
diff --git a/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.pyi b/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.pyi
index 239e2fbc1ac..9846b4e0e75 100644
--- a/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.pyi
+++ b/flyteidl/gen/pb_python/flyteidl/plugins/ray_pb2.pyi
@@ -1,5 +1,7 @@
 from flyteidl.core import tasks_pb2 as _tasks_pb2
+from flyteidl.core import literals_pb2 as _literals_pb2
 from google.protobuf.internal import containers as _containers
+from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper
 from google.protobuf import descriptor as _descriptor
 from google.protobuf import message as _message
 from typing import ClassVar as _ClassVar, Iterable as _Iterable, Mapping as _Mapping, Optional as _Optional, Union as _Union
@@ -20,15 +22,41 @@ class RayJob(_message.Message):
     runtime_env_yaml: str
     def __init__(self, ray_cluster: _Optional[_Union[RayCluster, _Mapping]] = ..., runtime_env: _Optional[str] = ..., shutdown_after_job_finishes: bool = ..., ttl_seconds_after_finished: _Optional[int] = ..., runtime_env_yaml: _Optional[str] = ...) -> None: ...
 
+class AutoscalerOptions(_message.Message):
+    __slots__ = ["upscaling_mode", "idle_timeout_seconds", "env", "image", "resources"]
+    class UpscalingMode(int, metaclass=_enum_type_wrapper.EnumTypeWrapper):
+        __slots__ = []
+        UPSCALING_MODE_UNSPECIFIED: _ClassVar[AutoscalerOptions.UpscalingMode]
+        UPSCALING_MODE_DEFAULT: _ClassVar[AutoscalerOptions.UpscalingMode]
+        UPSCALING_MODE_AGGRESSIVE: _ClassVar[AutoscalerOptions.UpscalingMode]
+        UPSCALING_MODE_CONSERVATIVE: _ClassVar[AutoscalerOptions.UpscalingMode]
+    UPSCALING_MODE_UNSPECIFIED: AutoscalerOptions.UpscalingMode
+    UPSCALING_MODE_DEFAULT: AutoscalerOptions.UpscalingMode
+    UPSCALING_MODE_AGGRESSIVE: AutoscalerOptions.UpscalingMode
+    UPSCALING_MODE_CONSERVATIVE: AutoscalerOptions.UpscalingMode
+    UPSCALING_MODE_FIELD_NUMBER: _ClassVar[int]
+    IDLE_TIMEOUT_SECONDS_FIELD_NUMBER: _ClassVar[int]
+    ENV_FIELD_NUMBER: _ClassVar[int]
+    IMAGE_FIELD_NUMBER: _ClassVar[int]
+    RESOURCES_FIELD_NUMBER: _ClassVar[int]
+    upscaling_mode: AutoscalerOptions.UpscalingMode
+    idle_timeout_seconds: int
+    env: _containers.RepeatedCompositeFieldContainer[_literals_pb2.KeyValuePair]
+    image: str
+    resources: _tasks_pb2.Resources
+    def __init__(self, upscaling_mode: _Optional[_Union[AutoscalerOptions.UpscalingMode, str]] = ..., idle_timeout_seconds: _Optional[int] = ..., env: _Optional[_Iterable[_Union[_literals_pb2.KeyValuePair, _Mapping]]] = ..., image: _Optional[str] = ..., resources: _Optional[_Union[_tasks_pb2.Resources, _Mapping]] = ...) -> None: ...
+
 class RayCluster(_message.Message):
-    __slots__ = ["head_group_spec", "worker_group_spec", "enable_autoscaling"]
+    __slots__ = ["head_group_spec", "worker_group_spec", "enable_autoscaling", "autoscaler_options"]
     HEAD_GROUP_SPEC_FIELD_NUMBER: _ClassVar[int]
     WORKER_GROUP_SPEC_FIELD_NUMBER: _ClassVar[int]
     ENABLE_AUTOSCALING_FIELD_NUMBER: _ClassVar[int]
+    AUTOSCALER_OPTIONS_FIELD_NUMBER: _ClassVar[int]
     head_group_spec: HeadGroupSpec
     worker_group_spec: _containers.RepeatedCompositeFieldContainer[WorkerGroupSpec]
     enable_autoscaling: bool
-    def __init__(self, head_group_spec: _Optional[_Union[HeadGroupSpec, _Mapping]] = ..., worker_group_spec: _Optional[_Iterable[_Union[WorkerGroupSpec, _Mapping]]] = ..., enable_autoscaling: bool = ...) -> None: ...
+    autoscaler_options: AutoscalerOptions
+    def __init__(self, head_group_spec: _Optional[_Union[HeadGroupSpec, _Mapping]] = ..., worker_group_spec: _Optional[_Iterable[_Union[WorkerGroupSpec, _Mapping]]] = ..., enable_autoscaling: bool = ..., autoscaler_options: _Optional[_Union[AutoscalerOptions, _Mapping]] = ...) -> None: ...
 
 class HeadGroupSpec(_message.Message):
     __slots__ = ["ray_start_params", "k8s_pod"]
diff --git a/flyteidl/gen/pb_rust/flyteidl.plugins.rs b/flyteidl/gen/pb_rust/flyteidl.plugins.rs
index 16589b3e60c..e118930748f 100644
--- a/flyteidl/gen/pb_rust/flyteidl.plugins.rs
+++ b/flyteidl/gen/pb_rust/flyteidl.plugins.rs
@@ -233,6 +233,58 @@ pub struct RayJob {
     #[prost(string, tag="5")]
     pub runtime_env_yaml: ::prost::alloc::string::String,
 }
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct AutoscalerOptions {
+    #[prost(enumeration="autoscaler_options::UpscalingMode", tag="1")]
+    pub upscaling_mode: i32,
+    #[prost(int32, tag="2")]
+    pub idle_timeout_seconds: i32,
+    /// autoscaler sidecar env vars
+    #[prost(message, repeated, tag="3")]
+    pub env: ::prost::alloc::vec::Vec<super::core::KeyValuePair>,
+    /// custom autoscaler image
+    #[prost(string, tag="4")]
+    pub image: ::prost::alloc::string::String,
+    /// autoscaler container resources
+    #[prost(message, optional, tag="5")]
+    pub resources: ::core::option::Option<super::core::Resources>,
+}
+/// Nested message and enum types in `AutoscalerOptions`.
+pub mod autoscaler_options {
+    #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+    #[repr(i32)]
+    pub enum UpscalingMode {
+        Unspecified = 0,
+        Default = 1,
+        Aggressive = 2,
+        Conservative = 3,
+    }
+    impl UpscalingMode {
+        /// String value of the enum field names used in the ProtoBuf definition.
+        ///
+        /// The values are not transformed in any way and thus are considered stable
+        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+        pub fn as_str_name(&self) -> &'static str {
+            match self {
+                UpscalingMode::Unspecified => "UPSCALING_MODE_UNSPECIFIED",
+                UpscalingMode::Default => "UPSCALING_MODE_DEFAULT",
+                UpscalingMode::Aggressive => "UPSCALING_MODE_AGGRESSIVE",
+                UpscalingMode::Conservative => "UPSCALING_MODE_CONSERVATIVE",
+            }
+        }
+        /// Creates an enum from field names used in the ProtoBuf definition.
+        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+            match value {
+                "UPSCALING_MODE_UNSPECIFIED" => Some(Self::Unspecified),
+                "UPSCALING_MODE_DEFAULT" => Some(Self::Default),
+                "UPSCALING_MODE_AGGRESSIVE" => Some(Self::Aggressive),
+                "UPSCALING_MODE_CONSERVATIVE" => Some(Self::Conservative),
+                _ => None,
+            }
+        }
+    }
+}
 /// Define Ray cluster defines the desired state of RayCluster
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -246,6 +298,8 @@ pub struct RayCluster {
     /// Whether to enable autoscaling.
     #[prost(bool, tag="3")]
     pub enable_autoscaling: bool,
+    #[prost(message, optional, tag="4")]
+    pub autoscaler_options: ::core::option::Option<AutoscalerOptions>,
 }
 /// HeadGroupSpec are the spec for the head pod
 #[allow(clippy::derive_partial_eq_without_eq)]
diff --git a/flyteidl/protos/flyteidl/plugins/ray.proto b/flyteidl/protos/flyteidl/plugins/ray.proto
index 749444ee049..d5c969f6873 100644
--- a/flyteidl/protos/flyteidl/plugins/ray.proto
+++ b/flyteidl/protos/flyteidl/plugins/ray.proto
@@ -5,6 +5,7 @@ package flyteidl.plugins;
 option go_package = "github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/plugins";
 
 import "flyteidl/core/tasks.proto";
+import "flyteidl/core/literals.proto";
 
 // RayJobSpec defines the desired state of RayJob
 message RayJob {
@@ -22,6 +23,23 @@ message RayJob {
   string runtime_env_yaml = 5;
 }
 
+message AutoscalerOptions {
+  enum UpscalingMode {
+    UPSCALING_MODE_UNSPECIFIED = 0;
+    UPSCALING_MODE_DEFAULT = 1;
+    UPSCALING_MODE_AGGRESSIVE = 2;
+    UPSCALING_MODE_CONSERVATIVE = 3;
+  }
+  UpscalingMode upscaling_mode = 1;
+  int32 idle_timeout_seconds = 2;
+  // autoscaler sidecar env vars
+  repeated core.KeyValuePair env = 3;
+  // custom autoscaler image
+  string image = 4;
+  // autoscaler container resources
+  core.Resources resources = 5;
+}
+
 // Define Ray cluster defines the desired state of RayCluster
 message RayCluster {
   // HeadGroupSpecs are the spec for the head pod
@@ -30,6 +48,7 @@ message RayCluster {
   repeated WorkerGroupSpec worker_group_spec = 2;
   // Whether to enable autoscaling.
   bool enable_autoscaling = 3;
+  AutoscalerOptions autoscaler_options = 4;
 }
 
 // HeadGroupSpec are the spec for the head pod
diff --git a/flyteplugins/go/tasks/plugins/k8s/ray/ray.go b/flyteplugins/go/tasks/plugins/k8s/ray/ray.go
index 1494d6463f8..0044038b2e4 100644
--- a/flyteplugins/go/tasks/plugins/k8s/ray/ray.go
+++ b/flyteplugins/go/tasks/plugins/k8s/ray/ray.go
@@ -13,6 +13,7 @@ import (
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	"gopkg.in/yaml.v2"
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/utils/ptr"
@@ -129,6 +130,55 @@ func (rayJobResourceHandler) BuildResource(ctx context.Context, taskCtx pluginsC
 	return rayjob, err
 }
 
+func buildAutoscalerOptions(options *plugins.AutoscalerOptions) *rayv1.AutoscalerOptions {
+	var autoScalerOptions *rayv1.AutoscalerOptions
+	if options != nil {
+		autoScalerOptions = &rayv1.AutoscalerOptions{}
+		if idleTimeoutTime := options.GetIdleTimeoutSeconds(); idleTimeoutTime > 0 {
+			autoScalerOptions.IdleTimeoutSeconds = &idleTimeoutTime
+		}
+		if upscalingMode := options.GetUpscalingMode(); upscalingMode != plugins.AutoscalerOptions_UPSCALING_MODE_UNSPECIFIED {
+			var mode rayv1.UpscalingMode
+			switch upscalingMode {
+			case plugins.AutoscalerOptions_UPSCALING_MODE_DEFAULT:
+				mode = rayv1.UpscalingMode("Default")
+			case plugins.AutoscalerOptions_UPSCALING_MODE_CONSERVATIVE:
+				mode = rayv1.UpscalingMode("Conservative")
+			case plugins.AutoscalerOptions_UPSCALING_MODE_AGGRESSIVE:
+				mode = rayv1.UpscalingMode("Aggressive")
+			}
+			autoScalerOptions.UpscalingMode = &mode
+		}
+		if image := options.GetImage(); image != "" {
+			autoScalerOptions.Image = &image
+		}
+		if res, err := flytek8s.ToK8sResourceRequirements(options.GetResources()); err == nil {
+			autoScalerOptions.Resources = res
+		}
+		autoScalerOptions.Env = flytek8s.ToK8sEnvVar(options.GetEnv())
+	}
+	return autoScalerOptions
+}
+
+func convertResourceEntriesToResourceList(entries []*core.Resources_ResourceEntry) v1.ResourceList {
+	resourceList := v1.ResourceList{}
+	for _, entry := range entries {
+		var name v1.ResourceName
+		switch entry.GetName() {
+		case core.Resources_CPU:
+			name = v1.ResourceCPU
+		case core.Resources_MEMORY:
+			name = v1.ResourceMemory
+		default:
+			continue
+		}
+		if q, err := resource.ParseQuantity(entry.GetValue()); err == nil {
+			resourceList[name] = q
+		}
+	}
+	return resourceList
+}
+
 func constructRayJob(taskCtx pluginsCore.TaskExecutionContext, rayJob *plugins.RayJob, objectMeta *metav1.ObjectMeta, taskPodSpec v1.PodSpec, headNodeRayStartParams map[string]string, primaryContainerIdx int, primaryContainer v1.Container) (*rayv1.RayJob, error) {
 	cfg := GetConfig()
 
@@ -155,6 +205,7 @@ func constructRayJob(taskCtx pluginsCore.TaskExecutionContext, rayJob *plugins.R
 		},
 		WorkerGroupSpecs:        []rayv1.WorkerGroupSpec{},
 		EnableInTreeAutoscaling: &rayJob.RayCluster.EnableAutoscaling,
+		AutoscalerOptions:       buildAutoscalerOptions(rayJob.GetRayCluster().GetAutoscalerOptions()),
 	}
 
 	for _, spec := range rayJob.GetRayCluster().GetWorkerGroupSpec() {
diff --git a/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go b/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go
index 1a268fc0e2e..0e191918ad8 100644
--- a/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go
+++ b/flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go
@@ -1592,3 +1592,135 @@ func transformStructToStructPB(t *testing.T, obj interface{}) *structpb.Struct {
 	assert.Nil(t, err)
 	return s
 }
+
+func TestConvertResourceEntriesToResourceList(t *testing.T) {
+	tests := []struct {
+		name     string
+		entries  []*core.Resources_ResourceEntry
+		wantKeys []corev1.ResourceName
+		wantLen  int
+	}{
+		{
+			name: "cpu entry",
+			entries: []*core.Resources_ResourceEntry{
+				{Name: core.Resources_CPU, Value: "500m"},
+			},
+			wantKeys: []corev1.ResourceName{corev1.ResourceCPU},
+			wantLen:  1,
+		},
+		{
+			name: "memory entry",
+			entries: []*core.Resources_ResourceEntry{
+				{Name: core.Resources_MEMORY, Value: "1Gi"},
+			},
+			wantKeys: []corev1.ResourceName{corev1.ResourceMemory},
+			wantLen:  1,
+		},
+		{
+			name: "unknown resource skipped",
+			entries: []*core.Resources_ResourceEntry{
+				{Name: core.Resources_ResourceName(99), Value: "1"},
+			},
+			wantLen: 0,
+		},
+		{
+			name: "invalid quantity skipped",
+			entries: []*core.Resources_ResourceEntry{
+				{Name: core.Resources_CPU, Value: "not-a-quantity"},
+			},
+			wantLen: 0,
+		},
+		{
+			name:    "empty input",
+			entries: []*core.Resources_ResourceEntry{},
+			wantLen: 0,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := convertResourceEntriesToResourceList(tt.entries)
+			require.NotNil(t, result)
+			assert.Len(t, result, tt.wantLen)
+			for _, key := range tt.wantKeys {
+				_, ok := result[key]
+				assert.True(t, ok, "expected key %s in result", key)
+			}
+		})
+	}
+}
+
+func TestBuildAutoscalerOptions(t *testing.T) {
+	t.Run("nil input returns nil", func(t *testing.T) {
+		assert.Nil(t, buildAutoscalerOptions(nil))
+	})
+
+	t.Run("idle timeout propagated", func(t *testing.T) {
+		result := buildAutoscalerOptions(&plugins.AutoscalerOptions{IdleTimeoutSeconds: 30})
+		require.NotNil(t, result)
+		require.NotNil(t, result.IdleTimeoutSeconds)
+		assert.Equal(t, int32(30), *result.IdleTimeoutSeconds)
+	})
+
+	t.Run("upscaling mode set when non-empty", func(t *testing.T) {
+		result := buildAutoscalerOptions(&plugins.AutoscalerOptions{UpscalingMode: plugins.AutoscalerOptions_UPSCALING_MODE_CONSERVATIVE})
+		require.NotNil(t, result)
+		require.NotNil(t, result.UpscalingMode)
+		assert.Equal(t, rayv1.UpscalingMode("Conservative"), *result.UpscalingMode)
+	})
+
+	t.Run("upscaling mode nil when empty", func(t *testing.T) {
+		result := buildAutoscalerOptions(&plugins.AutoscalerOptions{UpscalingMode: plugins.AutoscalerOptions_UPSCALING_MODE_UNSPECIFIED})
+		require.NotNil(t, result)
+		assert.Nil(t, result.UpscalingMode)
+	})
+
+	t.Run("image set when non-empty", func(t *testing.T) {
+		result := buildAutoscalerOptions(&plugins.AutoscalerOptions{Image: "my-image:latest"})
+		require.NotNil(t, result)
+		require.NotNil(t, result.Image)
+		assert.Equal(t, "my-image:latest", *result.Image)
+	})
+
+	t.Run("idle timeout zero should not be set", func(t *testing.T) {
+		result := buildAutoscalerOptions(&plugins.AutoscalerOptions{})
+		require.NotNil(t, result)
+		assert.Nil(t, result.IdleTimeoutSeconds)
+	})
+
+	t.Run("image nil when empty", func(t *testing.T) {
+		result := buildAutoscalerOptions(&plugins.AutoscalerOptions{Image: ""})
+		require.NotNil(t, result)
+		assert.Nil(t, result.Image)
+	})
+
+	t.Run("resources requests and limits converted", func(t *testing.T) {
+		result := buildAutoscalerOptions(&plugins.AutoscalerOptions{
+			Resources: &core.Resources{
+				Requests: []*core.Resources_ResourceEntry{
+					{Name: core.Resources_CPU, Value: "250m"},
+					{Name: core.Resources_MEMORY, Value: "512Mi"},
+				},
+				Limits: []*core.Resources_ResourceEntry{
+					{Name: core.Resources_CPU, Value: "1"},
+					{Name: core.Resources_MEMORY, Value: "1Gi"},
+				},
+			},
+		})
+		require.NotNil(t, result)
+		require.NotNil(t, result.Resources)
+		assert.Equal(t, resource.MustParse("250m"), result.Resources.Requests[corev1.ResourceCPU])
+		assert.Equal(t, resource.MustParse("512Mi"), result.Resources.Requests[corev1.ResourceMemory])
+		assert.Equal(t, resource.MustParse("1"), result.Resources.Limits[corev1.ResourceCPU])
+		assert.Equal(t, resource.MustParse("1Gi"), result.Resources.Limits[corev1.ResourceMemory])
+	})
+
+	t.Run("env literal value", func(t *testing.T) {
+		result := buildAutoscalerOptions(&plugins.AutoscalerOptions{
+			Env: []*core.KeyValuePair{{Key: "FOO", Value: "bar"}},
+		})
+		require.NotNil(t, result)
+		require.Len(t, result.Env, 1)
+		assert.Equal(t, "FOO", result.Env[0].Name)
+		assert.Equal(t, "bar", result.Env[0].Value)
+	})
+}