[6/N] feat(runway): add proto definitions, Ping controller, and gRPC server

BUILD.bazel

@@ -7,6 +7,7 @@ load("@gazelle//:def.bzl", "gazelle")
 # gazelle:exclude .claude
 
 # Resolve protobuf import ambiguities - use the actual protopb packages, not the proto aliases
+# gazelle:resolve go github.com/uber/submitqueue/runway/orchestrator/protopb //runway/orchestrator/protopb
 # gazelle:resolve go github.com/uber/submitqueue/submitqueue/gateway/protopb //submitqueue/gateway/protopb
 # gazelle:resolve go github.com/uber/submitqueue/submitqueue/orchestrator/protopb //submitqueue/orchestrator/protopb
 # gazelle:resolve go github.com/uber/submitqueue/stovepipe/gateway/protopb //stovepipe/gateway/protopb

Makefile

@@ -35,7 +35,7 @@ GOIMPORTS_VERSION ?= v0.33.0
 # Proto packages: <domain>/<service> dirs whose protopb/ holds the generated
 # stubs. Each is generated by Bazel into bazel-bin/tool/proto/<domain>_<service>/
 # (the out_dir convention in tool/proto/BUILD.bazel) and copied back here.
-PROTO_PACKAGES = submitqueue/gateway submitqueue/orchestrator stovepipe/gateway stovepipe/orchestrator
+PROTO_PACKAGES = runway/orchestrator submitqueue/gateway submitqueue/orchestrator stovepipe/gateway stovepipe/orchestrator
 
 # Set REPO_ROOT for docker-compose
 export REPO_ROOT := $(shell pwd)

example/runway/orchestrator/server/BUILD.bazel

@@ -15,10 +15,14 @@ go_library(
         "//runway/core/topickey",
         "//runway/extension/vcs",
         "//runway/extension/vcs/noop",
+        "//runway/orchestrator/controller",
         "//runway/orchestrator/controller/check",
         "//runway/orchestrator/controller/land",
+        "//runway/orchestrator/protopb",
         "@com_github_go_sql_driver_mysql//:mysql",
         "@com_github_uber_go_tally//:tally",
+        "@org_golang_google_grpc//:grpc",
+        "@org_golang_google_grpc//reflection",
         "@org_uber_go_zap//:zap",
     ],
 )

example/runway/orchestrator/server/main.go

@@ -19,6 +19,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"net"
 	"os"
 	"os/signal"
 	"sync"
@@ -28,6 +29,8 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/uber-go/tally"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
 
 	"github.com/uber/submitqueue/core/consumer"
 	"github.com/uber/submitqueue/core/errs"
@@ -38,8 +41,10 @@ import (
 	"github.com/uber/submitqueue/runway/core/topickey"
 	"github.com/uber/submitqueue/runway/extension/vcs"
 	"github.com/uber/submitqueue/runway/extension/vcs/noop"
+	"github.com/uber/submitqueue/runway/orchestrator/controller"
 	"github.com/uber/submitqueue/runway/orchestrator/controller/check"
 	"github.com/uber/submitqueue/runway/orchestrator/controller/land"
+	pb "github.com/uber/submitqueue/runway/orchestrator/protopb"
 )
 
 func main() {
@@ -56,6 +61,17 @@ func main() {
 	os.Exit(code)
 }
 
+// OrchestratorServer wraps the controller and implements the gRPC service interface.
+type OrchestratorServer struct {
+	pb.UnimplementedRunwayOrchestratorServer
+	pingController *controller.PingController
+}
+
+// Ping delegates to the controller.
+func (s *OrchestratorServer) Ping(ctx context.Context, req *pb.PingRequest) (*pb.PingResponse, error) {
+	return s.pingController.Ping(ctx, req)
+}
+
 // noopVCSFactory returns the noop VCS for any configuration.
 type noopVCSFactory struct{}
 
@@ -153,16 +169,48 @@ func run() error {
 		return fmt.Errorf("failed to start consumer: %w", err)
 	}
 
-	fmt.Println("Runway orchestrator is running. Press Ctrl+C to stop, or send a SIGTERM.")
+	// gRPC server for health checks (Ping).
+	grpcServer := grpc.NewServer()
+	pingController := controller.NewPingController(logger, scope)
+	srv := &OrchestratorServer{pingController: pingController}
+	pb.RegisterRunwayOrchestratorServer(grpcServer, srv)
+	reflection.Register(grpcServer)
+
+	port := os.Getenv("PORT")
+	if port == "" {
+		port = ":8085"
+	}
+	listener, err := net.Listen("tcp", port)
+	if err != nil {
+		return fmt.Errorf("failed to listen on port %s: %w", port, err)
+	}
+
+	fmt.Printf("Runway orchestrator is running on %s\n", port)
+	fmt.Println("Press Ctrl+C to stop, or send a SIGTERM.")
 
-	<-ctx.Done()
-	fmt.Println("Shutting down runway orchestrator...")
+	serverErrCh := make(chan error, 1)
+	go func() {
+		serverErrCh <- grpcServer.Serve(listener)
+	}()
 
-	err = ctx.Err()
+	var serverErr error
+	select {
+	case <-ctx.Done():
+		fmt.Println("Shutting down runway orchestrator...")
+		err = ctx.Err()
+		grpcServer.GracefulStop()
+		serverErr = <-serverErrCh
+	case serverErr = <-serverErrCh:
+		fmt.Println("Shutting down runway orchestrator due to gRPC server error...")
+	}
+
+	if serverErr != nil {
+		err = fmt.Errorf("gRPC server exited with error: %w", serverErr)
+	}
 
 	stopErr := primaryConsumer.Stop(30000)
 	if stopErr != nil {
-		err = fmt.Errorf("failed to stop consumer: %w", stopErr)
+		err = errors.Join(err, fmt.Errorf("failed to stop consumer: %w", stopErr))
 	}
 
 	return err

runway/orchestrator/controller/BUILD.bazel

@@ -0,0 +1,27 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "controller",
+    srcs = ["ping.go"],
+    importpath = "github.com/uber/submitqueue/runway/orchestrator/controller",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//core/metrics",
+        "//runway/orchestrator/protopb",
+        "@com_github_uber_go_tally//:tally",
+        "@org_uber_go_zap//:zap",
+    ],
+)
+
+go_test(
+    name = "controller_test",
+    srcs = ["ping_test.go"],
+    embed = [":controller"],
+    deps = [
+        "//runway/orchestrator/protopb",
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+        "@com_github_uber_go_tally//:tally",
+        "@org_uber_go_zap//:zap",
+    ],
+)

runway/orchestrator/controller/ping.go

@@ -0,0 +1,71 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controller
+
+import (
+	"context"
+	"os"
+	"time"
+
+	"github.com/uber-go/tally"
+	"github.com/uber/submitqueue/core/metrics"
+	pb "github.com/uber/submitqueue/runway/orchestrator/protopb"
+	"go.uber.org/zap"
+)
+
+// PingController handles ping business logic for the Runway orchestrator.
+type PingController struct {
+	logger       *zap.Logger
+	metricsScope tally.Scope
+}
+
+// NewPingController creates a new instance of the Runway orchestrator ping controller.
+func NewPingController(logger *zap.Logger, scope tally.Scope) *PingController {
+	return &PingController{
+		logger:       logger,
+		metricsScope: scope,
+	}
+}
+
+// Ping handles the ping request and returns a response.
+func (c *PingController) Ping(ctx context.Context, req *pb.PingRequest) (resp *pb.PingResponse, retErr error) {
+	const opName = "ping"
+
+	op := metrics.Begin(c.metricsScope, opName)
+	defer func() { op.Complete(retErr) }()
+
+	message := "pong!"
+	isEcho := false
+	if req.Message != "" {
+		message = "echo: " + req.Message
+		isEcho = true
+		metrics.NamedCounter(c.metricsScope, opName, "echo_requests", 1)
+	}
+
+	hostname, _ := os.Hostname()
+
+	c.logger.Info("ping request received",
+		zap.String("message", req.Message),
+		zap.Bool("is_echo", isEcho),
+		zap.String("hostname", hostname),
+	)
+
+	return &pb.PingResponse{
+		Message:     message,
+		ServiceName: "runway-orchestrator",
+		Timestamp:   time.Now().Unix(),
+		Hostname:    hostname,
+	}, nil
+}

runway/orchestrator/controller/ping_test.go

@@ -0,0 +1,68 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package controller
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+	pb "github.com/uber/submitqueue/runway/orchestrator/protopb"
+	"go.uber.org/zap"
+)
+
+func TestNewPingController(t *testing.T) {
+	ctrl := NewPingController(zap.NewNop(), tally.NoopScope)
+	require.NotNil(t, ctrl)
+}
+
+func TestPing_DefaultMessage(t *testing.T) {
+	ctrl := NewPingController(zap.NewNop(), tally.NoopScope)
+	ctx := context.Background()
+
+	req := &pb.PingRequest{}
+	resp, err := ctrl.Ping(ctx, req)
+
+	require.NoError(t, err)
+	assert.Equal(t, "pong!", resp.Message)
+}
+
+func TestPing_ServiceName(t *testing.T) {
+	ctrl := NewPingController(zap.NewNop(), tally.NoopScope)
+	ctx := context.Background()
+
+	req := &pb.PingRequest{}
+	resp, err := ctrl.Ping(ctx, req)
+
+	require.NoError(t, err)
+	assert.Equal(t, "runway-orchestrator", resp.ServiceName)
+}
+
+func TestPing_Timestamp(t *testing.T) {
+	ctrl := NewPingController(zap.NewNop(), tally.NoopScope)
+	ctx := context.Background()
+
+	before := time.Now().Unix()
+	req := &pb.PingRequest{}
+	resp, err := ctrl.Ping(ctx, req)
+	after := time.Now().Unix()
+
+	require.NoError(t, err)
+	assert.GreaterOrEqual(t, resp.Timestamp, before)
+	assert.LessOrEqual(t, resp.Timestamp, after)
+}

runway/orchestrator/proto/BUILD.bazel

@@ -0,0 +1,40 @@
+load("@rules_go//go:def.bzl", "go_library")
+load("@rules_go//proto:def.bzl", "go_proto_library")
+load("@rules_proto//proto:defs.bzl", "proto_library")
+
+exports_files(
+    ["orchestrator.proto"],
+    visibility = ["//tool/proto:__pkg__"],
+)
+
+proto_library(
+    name = "orchestratorpb_proto",
+    srcs = ["orchestrator.proto"],
+    visibility = ["//visibility:public"],
+)
+
+# keep
+go_proto_library(
+    name = "orchestratorpb_go_proto",
+    compilers = [
+        "@rules_go//proto:go_proto",
+        "@rules_go//proto:go_grpc_v2",
+    ],
+    importpath = "github.com/uber/submitqueue/runway/orchestrator/proto",
+    proto = ":orchestratorpb_proto",
+    visibility = ["//visibility:public"],
+)
+
+go_library(
+    name = "proto",
+    embed = [":orchestratorpb_go_proto"],
+    importpath = "github.com/uber/submitqueue/runway/orchestrator/proto",
+    visibility = ["//visibility:public"],
+)
+
+go_library(
+    name = "protopb",
+    embed = [":orchestratorpb_go_proto"],
+    importpath = "github.com/uber/submitqueue/runway/orchestrator/protopb",
+    visibility = ["//visibility:public"],
+)

runway/orchestrator/proto/orchestrator.proto

@@ -0,0 +1,46 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package uber.submitqueue.runway.orchestrator;
+
+option go_package = "github.com/uber/submitqueue/runway/orchestrator/protopb";
+option java_multiple_files = true;
+option java_outer_classname = "OrchestratorProto";
+option java_package = "com.uber.submitqueue.runway.orchestrator";
+
+// PingRequest is the request for the Ping method
+message PingRequest {
+    // Optional message to include in the ping
+    string message = 1;
+}
+
+// PingResponse is the response for the Ping method
+message PingResponse {
+    // The response message
+    string message = 1;
+    // The service name that handled the request
+    string service_name = 2;
+    // Timestamp of when the ping was received
+    int64 timestamp = 3;
+    // Hostname of the server that handled the request
+    string hostname = 4;
+}
+
+// RunwayOrchestrator provides the Runway orchestrator API.
+service RunwayOrchestrator {
+    // Ping returns a response indicating the service is alive
+    rpc Ping(PingRequest) returns (PingResponse) {}
+}