[ACTP] add par-control executor channel

Author: embeaken

Cargo.lock

@@ -41,7 +41,9 @@ protoc-gen-prost = "0.5"
 protoc-gen-tonic = "0.5"
 lru = { version = "0.18.0", default-features = false }
 memchr = "2.7.6"
+native-tls = { version = "0.2", features = ["alpn"] }
 nom = "8.0"
+openssl = "0.10"
 normalize-path = "0.2"
 phf = { version = "0.14", features = ["macros"] }
 rawzip = "0.5.0"
@@ -57,6 +59,7 @@ time = { version = "0.3", features = ["formatting", "macros"] }
 thiserror = "2.0.12"
 tokio = "1"
 tokio-stream = "0.1"
+tokio-native-tls = "0.3"
 tonic = "0.14"
 tonic-build = "0.14"
 tonic-prost = "0.14"

Cargo.toml

@@ -276,7 +276,7 @@ func (p *PrivateActionRunner) StartExecutor(ctx context.Context) error {
 }
 
 func (p *PrivateActionRunner) startExecutor(ctx context.Context) error {
-	// Detached from ctx's deadline: the server must run until Stop(), not until the fx start timeout.
+	// The server runs until Stop, independently of the Fx startup deadline.
 	runCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
 	p.cancelStart = cancel
 	defer p.logger.Flush()
@@ -304,12 +304,14 @@ func (p *PrivateActionRunner) startExecutor(ctx context.Context) error {
 	p.encryptionStore = encryptioncontext.NewStore()
 	taskExecutor := runners.NewWorkflowTaskExecutor(cfg, taskVerifier, p.traceroute, p.eventPlatform, p.ipc.GetClient(), p.encryptionStore, p.ha, p.ka)
 
-	p.executorServer = executor.NewServer(taskExecutor, parversion.RunnerVersion)
+	p.executorServer = executor.NewServer(taskExecutor, parversion.RunnerVersion, keysManager)
 
 	go p.encryptionStore.Start()
 	keysManager.Start(runCtx)
 	go func() {
-		keysManager.WaitForReady()
+		if err := keysManager.WaitForReady(runCtx); err != nil {
+			return
+		}
 		p.executorServer.SetReady(true)
 		p.logger.Info("Private action runner executor ready to accept actions")
 	}()

MODULE.bazel.lock

@@ -20,4 +20,20 @@ crate.annotation(
     crate = "protoc-gen-tonic",
     gen_binaries = ["protoc-gen-tonic"],
 )
+
+# Link openssl-sys against the Agent's packaged OpenSSL.
+crate.annotation(
+    build_script_data = ["@openssl//:gen_dir"],
+    build_script_env = {
+        "OPENSSL_NO_VENDOR": "1",
+        "OPENSSL_DIR": "$(execpath @openssl//:gen_dir)",
+    },
+    crate = "openssl-sys",
+    data = ["@openssl"],
+    deps = ["@openssl"],
+)
+
+# Expose @openssl to the crate extension used above.
+inject_repo(crate, "openssl")
+
 use_repo(crate, "crates")

comp/privateactionrunner/impl/privateactionrunner.go

@@ -218,6 +218,14 @@ configure_make(
     }),
 )
 
+# The full install tree (include/ + lib/) as a single directory, for consumers
+# that need an OPENSSL_DIR-style prefix (e.g. the Rust openssl-sys build script).
+filegroup(
+    name = "gen_dir",
+    srcs = [":openssl"],
+    output_group = "gen_dir",
+)
+
 foreign_cc_shared_wrapper(
     name = "openssl_shared",
     input = ":openssl",

deps/crates.MODULE.bazel

@@ -13,12 +13,15 @@ go_library(
     deps = [
         "//pkg/privateactionrunner/adapters/logging",
         "//pkg/privateactionrunner/runners",
+        "//pkg/privateactionrunner/task-verifier",
         "//pkg/privateactionrunner/types",
         "//pkg/privateactionrunner/util",
         "//pkg/proto/pbgo/privateactionrunner/errorcode",
         "//pkg/proto/pbgo/privateactionrunner/executor",
         "@com_github_benbjohnson_clock//:clock",
         "@org_golang_google_grpc//:grpc",
+        "@org_golang_google_grpc//codes",
+        "@org_golang_google_grpc//status",
     ] + select({
         "@rules_go//go/platform:windows": [
             "@com_github_microsoft_go_winio//:go-winio",
@@ -37,6 +40,7 @@ dd_agent_go_test(
     embed = [":executor"],
     deps = [
         "//pkg/privateactionrunner/runners",
+        "//pkg/privateactionrunner/task-verifier",
         "//pkg/privateactionrunner/types",
         "//pkg/privateactionrunner/util",
         "//pkg/proto/pbgo/privateactionrunner/errorcode",
@@ -45,7 +49,10 @@ dd_agent_go_test(
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
         "@org_golang_google_grpc//:grpc",
+        "@org_golang_google_grpc//codes",
         "@org_golang_google_grpc//credentials",
         "@org_golang_google_grpc//credentials/insecure",
+        "@org_golang_google_grpc//status",
+        "@org_golang_google_protobuf//proto",
     ],
 )