From 4b219f22985033bd7b71f8a73ca43089e56e7c92 Mon Sep 17 00:00:00 2001 From: Roy Larsen Date: Thu, 30 Jul 2026 11:47:39 -0400 Subject: [PATCH 1/5] scaleset SDK spike --- .github/workflows/ci.yml | 2 +- Dockerfile | 2 +- Makefile | 4 +- go.mod | 22 +- go.sum | 61 +++--- main.go | 11 +- pkg/env/constants.go | 2 + pkg/env/env.go | 19 ++ pkg/github/scalesetclient/assert.go | 5 + pkg/github/scalesetclient/client.go | 268 +++++++++++++++++++++++ pkg/github/scalesetclient/client_test.go | 210 ++++++++++++++++++ pkg/github/scalesetclient/config_test.go | 29 +++ pkg/github/scalesetclient/stub_test.go | 248 +++++++++++++++++++++ pkg/metrics/metrics.go | 4 +- 14 files changed, 838 insertions(+), 49 deletions(-) create mode 100644 pkg/github/scalesetclient/assert.go create mode 100644 pkg/github/scalesetclient/client.go create mode 100644 pkg/github/scalesetclient/client_test.go create mode 100644 pkg/github/scalesetclient/config_test.go create mode 100644 pkg/github/scalesetclient/stub_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e4bd242..7054f12 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -16,7 +16,7 @@ on: required: false env: - GOLANG_VERSION: "1.23.0" + GOLANG_VERSION: "1.25.3" GHCR_REPO: ghcr.io/macstadium/orka-github-runner jobs: diff --git a/Dockerfile b/Dockerfile index 79c3c85..356b57f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.23.0 as builder +FROM golang:1.25.3 as builder WORKDIR /workspace diff --git a/Makefile b/Makefile index 4415847..430d843 100644 --- a/Makefile +++ b/Makefile @@ -17,9 +17,9 @@ all: tidy fmt lint build .PHONY: lint lint: @echo "[golangci-lint] Running golangci-lint..." - @go run github.com/golangci/golangci-lint/cmd/golangci-lint@v1.60.3 run --timeout=5m 2>&1 + @go run github.com/golangci/golangci-lint/cmd/golangci-lint@v1.64.8 run --timeout=5m 2>&1 @echo "[staticcheck] Running staticcheck..." - @go run honnef.co/go/tools/cmd/staticcheck@v0.5.1 + @go run honnef.co/go/tools/cmd/staticcheck@v0.7.0 @echo "------------------------------------[Done]" .PHONY: fmt diff --git a/go.mod b/go.mod index 5b8f085..e3cbd99 100644 --- a/go.mod +++ b/go.mod @@ -1,26 +1,28 @@ module github.com/macstadium/orka-github-actions-integration -go 1.23.0 +go 1.25.3 require ( + github.com/actions/scaleset v0.4.0 github.com/cenkalti/backoff/v4 v4.3.0 - github.com/golang-jwt/jwt/v4 v4.5.0 + github.com/golang-jwt/jwt/v4 v4.5.2 github.com/google/uuid v1.6.0 - github.com/hashicorp/go-retryablehttp v0.7.5 + github.com/hashicorp/go-retryablehttp v0.7.8 github.com/hashicorp/go-version v1.6.0 github.com/joho/godotenv v1.5.1 github.com/onsi/ginkgo/v2 v2.15.0 github.com/onsi/gomega v1.30.0 github.com/prometheus/client_golang v1.23.2 go.uber.org/zap v1.26.0 - golang.org/x/crypto v0.41.0 + go.uber.org/zap/exp v0.3.0 + golang.org/x/crypto v0.49.0 k8s.io/apimachinery v0.27.4 ) require ( github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect - github.com/go-logr/logr v1.3.0 // indirect + github.com/go-logr/logr v1.4.3 // indirect github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect github.com/google/go-cmp v0.7.0 // indirect github.com/google/pprof v0.0.0-20210720184732-4bb14d4b1be1 // indirect @@ -32,11 +34,11 @@ require ( github.com/rogpeppe/go-internal v1.13.1 // indirect go.uber.org/multierr v1.11.0 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect - golang.org/x/net v0.43.0 // indirect - golang.org/x/sys v0.35.0 // indirect - golang.org/x/text v0.28.0 // indirect - golang.org/x/tools v0.35.0 // indirect - google.golang.org/protobuf v1.36.8 // indirect + golang.org/x/net v0.52.0 // indirect + golang.org/x/sys v0.42.0 // indirect + golang.org/x/text v0.35.0 // indirect + golang.org/x/tools v0.42.0 // indirect + google.golang.org/protobuf v1.36.10 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect k8s.io/utils v0.0.0-20230209194617-a36077c30491 // indirect ) diff --git a/go.sum b/go.sum index 65868c9..4b15178 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/actions/scaleset v0.4.0 h1:691GC2AkHb3ZGjfNvatboYoRS7CLr3+4VcZk/6w9IbM= +github.com/actions/scaleset v0.4.0/go.mod h1:2L2I6rggFWV+zprDet6y7y7Vkm3HPudaup78eSc79Uo= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= @@ -8,14 +10,17 @@ github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWR github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI= github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/go-logr/logr v1.3.0 h1:2y3SDp0ZXuc6/cjLSZ+Q3ir+QB9T/iG5yYRXqsagWSY= -github.com/go-logr/logr v1.3.0/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM= +github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 h1:tfuBGBXKqDEevZMzYi5KSi8KkcZtzBcTgAUUtapy0OI= github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572/go.mod h1:9Pwr4B2jHnOSGXyyzV8ROjYa2ojvAY6HCGYYfMoC3Ls= -github.com/golang-jwt/jwt/v4 v4.5.0 h1:7cYmW1XlMY7h7ii7UhUyChSgS5wUJEnm9uZVTGqOWzg= -github.com/golang-jwt/jwt/v4 v4.5.0/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= +github.com/golang-jwt/jwt/v4 v4.5.2 h1:YtQM7lnr8iZ+j5q71MGKkNw9Mn7AjHM68uc9g5fXeUI= +github.com/golang-jwt/jwt/v4 v4.5.2/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= @@ -26,10 +31,10 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ= github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48= -github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI= -github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ= -github.com/hashicorp/go-retryablehttp v0.7.5 h1:bJj+Pj19UZMIweq/iie+1u5YCdGrnxCT9yvm0e+Nd5M= -github.com/hashicorp/go-retryablehttp v0.7.5/go.mod h1:Jy/gPYAdjqffZ/yFGCFV2doI5wjtH1ewM9u8iYVjtX8= +github.com/hashicorp/go-hclog v1.6.3 h1:Qr2kF+eVWjTiYmU7Y31tYlP1h0q/X3Nl3tPGdaB11/k= +github.com/hashicorp/go-hclog v1.6.3/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= +github.com/hashicorp/go-retryablehttp v0.7.8 h1:ylXZWnqa7Lhqpk0L1P1LzDtGcCR0rPVUrx/c8Unxc48= +github.com/hashicorp/go-retryablehttp v0.7.8/go.mod h1:rjiScheydd+CxvumBsIrFKlx3iS0jrZ7LvzFGFmuKbw= github.com/hashicorp/go-version v1.6.0 h1:feTTfFNnjP967rlCxM/I9g701jU+RN74YKx2mOkIeek= github.com/hashicorp/go-version v1.6.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc= @@ -43,14 +48,19 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= +github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= +github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/onsi/ginkgo/v2 v2.15.0 h1:79HwNRBAZHOEwrczrgSOPy+eFTTlIGELKy5as+ClttY= github.com/onsi/ginkgo/v2 v2.15.0/go.mod h1:HlxMHtYF57y6Dpf+mc5529KKmSq9h2FpCF+/ZkwUxKM= github.com/onsi/gomega v1.30.0 h1:hvMK7xYz4D3HapigLTeGdId/NcfQx1VHMJc60ew99+8= github.com/onsi/gomega v1.30.0/go.mod h1:9sxs+SwGrKI0+PWe4Fxa9tFQQBG5xSsSbMXOI8PPpoQ= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= @@ -62,7 +72,6 @@ github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlT github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= -github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= @@ -72,23 +81,25 @@ go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo= go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so= +go.uber.org/zap/exp v0.3.0 h1:6JYzdifzYkGmTdRR59oYH+Ng7k49H9qVpWwNSsGJj3U= +go.uber.org/zap/exp v0.3.0/go.mod h1:5I384qq7XGxYyByIhHm6jg5CHkGY0nsTfbDLgDDlgJQ= go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= -golang.org/x/crypto v0.41.0 h1:WKYxWedPGCTVVl5+WHSSrOBT0O8lx32+zxmHxijgXp4= -golang.org/x/crypto v0.41.0/go.mod h1:pO5AFd7FA68rFak7rOAGVuygIISepHftHnr8dr6+sUc= -golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= -golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= +golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= +golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA= +golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= +golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= -golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= -golang.org/x/term v0.34.0 h1:O/2T7POpk0ZZ7MAzMeWFSg6S5IpWd/RXDlM9hgM3DR4= -golang.org/x/term v0.34.0/go.mod h1:5jC53AEywhIVebHgPVeg0mj8OD3VO9OzclacVrqpaAw= -golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng= -golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= -golang.org/x/tools v0.35.0 h1:mBffYraMEf7aa0sB+NuKnuCy8qI/9Bughn8dC2Gu5r0= -golang.org/x/tools v0.35.0/go.mod h1:NKdj5HkL/73byiZSJjqJgKn3ep7KjFkBOkR/Hps3VPw= -google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= -google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= +golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= +golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.41.0 h1:QCgPso/Q3RTJx2Th4bDLqML4W6iJiaXFq2/ftQF13YU= +golang.org/x/term v0.41.0/go.mod h1:3pfBgksrReYfZ5lvYM0kSO0LIkAl4Yl2bXOkKP7Ec2A= +golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= +golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= +golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= +golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/main.go b/main.go index 30d1c19..e091e6f 100644 --- a/main.go +++ b/main.go @@ -11,9 +11,9 @@ import ( "github.com/macstadium/orka-github-actions-integration/pkg/constants" "github.com/macstadium/orka-github-actions-integration/pkg/env" - "github.com/macstadium/orka-github-actions-integration/pkg/github" "github.com/macstadium/orka-github-actions-integration/pkg/github/actions" "github.com/macstadium/orka-github-actions-integration/pkg/github/runners" + "github.com/macstadium/orka-github-actions-integration/pkg/github/scalesetclient" "github.com/macstadium/orka-github-actions-integration/pkg/github/types" "github.com/macstadium/orka-github-actions-integration/pkg/logging" "github.com/macstadium/orka-github-actions-integration/pkg/metrics" @@ -33,11 +33,6 @@ func main() { logging.SetupLogger(envData.LogLevel) logger := logging.Logger.Named("main") - config, err := github.NewGitHubConfig(envData.GitHubURL) - if err != nil { - panic(err) - } - runnerName := envData.Runners[0].Name groupId := constants.DefaultRunnerGroupID if envData.Runners[0].Id != 0 { @@ -48,7 +43,7 @@ func main() { panic(fmt.Sprintf("invalid runner name: %s. Runner name must consist of lower case alphanumeric characters or ' - ', start with an alphabetic character, end with an alphanumeric character, and may not be longer than 63 characters.", runnerName)) } - actionsClient, err := actions.NewActionsClient(ctx, envData, config) + actionsClient, err := scalesetclient.New(envData, envData.MaxRunners) if err != nil { panic(err) } @@ -150,7 +145,7 @@ func main() { run(ctx, runnerMessageProcessor, runnerScaleSet, logger) } -func createScaleSet(ctx context.Context, actionsClient *actions.ActionsClient, runnerName string, groupId int) (*types.RunnerScaleSet, error) { +func createScaleSet(ctx context.Context, actionsClient actions.ActionsService, runnerName string, groupId int) (*types.RunnerScaleSet, error) { return actionsClient.CreateRunnerScaleSet(ctx, &types.RunnerScaleSet{ Name: runnerName, RunnerGroupId: groupId, diff --git a/pkg/env/constants.go b/pkg/env/constants.go index 5c1f771..02309c4 100644 --- a/pkg/env/constants.go +++ b/pkg/env/constants.go @@ -29,6 +29,8 @@ const ( VMTrackerIntervalEnvName = "VM_TRACKER_INTERVAL" + MaxRunnersEnvName = "MAX_RUNNERS" + LogLevelEnvName = "LOG_LEVEL" // Prometheus metrics diff --git a/pkg/env/env.go b/pkg/env/env.go index 7b60941..e045290 100644 --- a/pkg/env/env.go +++ b/pkg/env/env.go @@ -50,6 +50,8 @@ type Data struct { VMTrackerInterval time.Duration + MaxRunners int + LogLevel string EnableMetrics bool @@ -89,6 +91,8 @@ func ParseEnv() *Data { VMTrackerInterval: getDurationEnv(VMTrackerIntervalEnvName, 300*time.Second), + MaxRunners: getIntEnv(MaxRunnersEnvName, 10), + LogLevel: getEnvWithDefault(LogLevelEnvName, logging.LogLevelInfo), EnableMetrics: getBoolEnv(EnableMetricsEnvName, false), @@ -217,6 +221,21 @@ func getBoolEnv(key string, fallback bool) bool { return strings.ToLower(value) == "true" } +func getIntEnv(key string, fallback int) int { + value := os.Getenv(key) + + if len(value) == 0 { + return fallback + } + + parsed, err := strconv.Atoi(value) + if err != nil { + return fallback + } + + return parsed +} + func getDurationEnv(key string, fallback time.Duration) time.Duration { value := os.Getenv(key) diff --git a/pkg/github/scalesetclient/assert.go b/pkg/github/scalesetclient/assert.go new file mode 100644 index 0000000..618279a --- /dev/null +++ b/pkg/github/scalesetclient/assert.go @@ -0,0 +1,5 @@ +package scalesetclient + +import "github.com/macstadium/orka-github-actions-integration/pkg/github/actions" + +var _ actions.ActionsService = (*Client)(nil) diff --git a/pkg/github/scalesetclient/client.go b/pkg/github/scalesetclient/client.go new file mode 100644 index 0000000..4ba4484 --- /dev/null +++ b/pkg/github/scalesetclient/client.go @@ -0,0 +1,268 @@ +package scalesetclient + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "sync" + + "github.com/actions/scaleset" + "github.com/google/uuid" + "github.com/macstadium/orka-github-actions-integration/pkg/env" + "github.com/macstadium/orka-github-actions-integration/pkg/github/types" + "github.com/macstadium/orka-github-actions-integration/pkg/logging" + "go.uber.org/zap/exp/zapslog" +) + +const jobMessagesType = "RunnerScaleSetJobMessages" + +type Client struct { + sdk *scaleset.Client + logger *slog.Logger + + mu sync.Mutex + session *scaleset.MessageSessionClient + maxRunners int +} + +func New(envData *env.Data, maxRunners int) (*Client, error) { + logger := slog.New(zapslog.NewHandler(logging.Logger.Named("scaleset").Desugar().Core())) + + sdk, err := scaleset.NewClientWithGitHubApp(scaleset.ClientWithGitHubAppConfig{ + GitHubConfigURL: envData.GitHubURL, + GitHubAppAuth: scaleset.GitHubAppAuth{ + ClientID: fmt.Sprintf("%d", envData.GitHubAppID), + InstallationID: envData.GitHubAppInstallationID, + PrivateKey: envData.GitHubAppPrivateKey, + }, + SystemInfo: scaleset.SystemInfo{ + System: "orka-github-actions-integration", + Subsystem: "listener", + }, + }, scaleset.WithLogger(logger)) + if err != nil { + return nil, fmt.Errorf("failed to create scaleset client: %w", err) + } + + return &Client{sdk: sdk, logger: logger, maxRunners: maxRunners}, nil +} + +func (c *Client) GetRunnerScaleSet(ctx context.Context, runnerGroupId int, runnerScaleSetName string) (*types.RunnerScaleSet, error) { + got, err := c.sdk.GetRunnerScaleSet(ctx, runnerGroupId, runnerScaleSetName) + if err != nil || got == nil { + return nil, err + } + return toScaleSet(got), nil +} + +func (c *Client) CreateRunnerScaleSet(ctx context.Context, runnerScaleSet *types.RunnerScaleSet) (*types.RunnerScaleSet, error) { + created, err := c.sdk.CreateRunnerScaleSet(ctx, fromScaleSet(runnerScaleSet)) + if err != nil || created == nil { + return nil, err + } + return toScaleSet(created), nil +} + +func (c *Client) DeleteRunnerScaleSet(ctx context.Context, runnerScaleSetId int) error { + return c.sdk.DeleteRunnerScaleSet(ctx, runnerScaleSetId) +} + +func (c *Client) GetRunner(ctx context.Context, runnerName string) (*types.RunnerReference, error) { + got, err := c.sdk.GetRunnerByName(ctx, runnerName) + if err != nil || got == nil { + return nil, err + } + return &types.RunnerReference{Id: got.ID, Name: got.Name, RunnerScaleSetId: got.RunnerScaleSetID}, nil +} + +func (c *Client) CreateRunner(ctx context.Context, runnerScaleSetID int, runnerName string) (*types.RunnerScaleSetJitRunnerConfig, error) { + cfg, err := c.sdk.GenerateJitRunnerConfig(ctx, &scaleset.RunnerScaleSetJitRunnerSetting{Name: runnerName}, runnerScaleSetID) + if err != nil || cfg == nil { + return nil, err + } + + out := &types.RunnerScaleSetJitRunnerConfig{EncodedJITConfig: cfg.EncodedJITConfig} + if cfg.Runner != nil { + out.Runner = &types.RunnerReference{Id: cfg.Runner.ID, Name: cfg.Runner.Name, RunnerScaleSetId: cfg.Runner.RunnerScaleSetID} + } + return out, nil +} + +func (c *Client) DeleteRunner(ctx context.Context, runnerID int) error { + return c.sdk.RemoveRunner(ctx, int64(runnerID)) +} + +func (c *Client) CreateMessageSession(ctx context.Context, runnerScaleSetId int, owner string) (*types.RunnerScaleSetSession, error) { + sessionClient, err := c.sdk.MessageSessionClient(ctx, runnerScaleSetId, owner) + if err != nil { + return nil, err + } + + c.mu.Lock() + c.session = sessionClient + c.mu.Unlock() + + return toSession(sessionClient.Session()), nil +} + +func (c *Client) RefreshMessageSession(ctx context.Context, runnerScaleSetId int, sessionId *uuid.UUID) (*types.RunnerScaleSetSession, error) { + session, err := c.currentSession() + if err != nil { + return nil, err + } + return toSession(session.Session()), nil +} + +func (c *Client) DeleteMessageSession(ctx context.Context, runnerScaleSetId int, sessionId *uuid.UUID) error { + session, err := c.currentSession() + if err != nil { + return err + } + return session.Close(ctx) +} + +func (c *Client) AcquireJobs(ctx context.Context, runnerScaleSetId int, messageQueueAccessToken string, requestIds []int64) ([]int64, error) { + session, err := c.currentSession() + if err != nil { + return nil, err + } + return session.AcquireJobs(ctx, requestIds) +} + +func (c *Client) GetMessage(ctx context.Context, messageQueueUrl, messageQueueAccessToken string, lastMessageId int64) (*types.RunnerScaleSetMessage, error) { + session, err := c.currentSession() + if err != nil { + return nil, err + } + + msg, err := session.GetMessage(ctx, int(lastMessageId), c.maxRunners) + if err != nil || msg == nil { + return nil, err + } + + body, err := encodeBody(msg) + if err != nil { + return nil, err + } + + return &types.RunnerScaleSetMessage{ + MessageId: int64(msg.MessageID), + MessageType: jobMessagesType, + Body: body, + Statistics: toStatistics(msg.Statistics), + }, nil +} + +func (c *Client) DeleteMessage(ctx context.Context, messageQueueUrl, messageQueueAccessToken string, messageId int64) error { + session, err := c.currentSession() + if err != nil { + return err + } + return session.DeleteMessage(ctx, int(messageId)) +} + +func (c *Client) GetAcquirableJobs(ctx context.Context, runnerScaleSetId int) (*types.AcquirableJobList, error) { + return &types.AcquirableJobList{Count: 0, Jobs: []types.AcquirableJob{}}, nil +} + +func (c *Client) currentSession() (*scaleset.MessageSessionClient, error) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.session == nil { + return nil, fmt.Errorf("no message session has been created") + } + return c.session, nil +} + +func encodeBody(msg *scaleset.RunnerScaleSetMessage) (string, error) { + batch := make([]any, 0, len(msg.JobAvailableMessages)+len(msg.JobAssignedMessages)+len(msg.JobStartedMessages)+len(msg.JobCompletedMessages)) + for _, m := range msg.JobAvailableMessages { + batch = append(batch, m) + } + for _, m := range msg.JobAssignedMessages { + batch = append(batch, m) + } + for _, m := range msg.JobStartedMessages { + batch = append(batch, m) + } + for _, m := range msg.JobCompletedMessages { + batch = append(batch, m) + } + + if len(batch) == 0 { + return "", nil + } + + encoded, err := json.Marshal(batch) + if err != nil { + return "", fmt.Errorf("failed to encode job messages: %w", err) + } + return string(encoded), nil +} + +func toScaleSet(in *scaleset.RunnerScaleSet) *types.RunnerScaleSet { + labels := make([]types.RunnerScaleSetLabel, 0, len(in.Labels)) + for _, l := range in.Labels { + labels = append(labels, types.RunnerScaleSetLabel{Type: l.Type, Name: l.Name}) + } + + return &types.RunnerScaleSet{ + Id: in.ID, + Name: in.Name, + RunnerGroupId: in.RunnerGroupID, + RunnerGroupName: in.RunnerGroupName, + Labels: labels, + RunnerSetting: types.RunnerScaleSetSetting{DisableUpdate: in.RunnerSetting.DisableUpdate}, + CreatedOn: in.CreatedOn, + RunnerJitConfigUrl: in.RunnerJitConfigURL, + Statistics: toStatistics(in.Statistics), + } +} + +func fromScaleSet(in *types.RunnerScaleSet) *scaleset.RunnerScaleSet { + labels := make([]scaleset.Label, 0, len(in.Labels)) + for _, l := range in.Labels { + labels = append(labels, scaleset.Label{Type: l.Type, Name: l.Name}) + } + + return &scaleset.RunnerScaleSet{ + ID: in.Id, + Name: in.Name, + RunnerGroupID: in.RunnerGroupId, + RunnerGroupName: in.RunnerGroupName, + Labels: labels, + RunnerSetting: scaleset.RunnerSetting{DisableUpdate: in.RunnerSetting.DisableUpdate}, + } +} + +func toSession(in scaleset.RunnerScaleSetSession) *types.RunnerScaleSetSession { + sessionId := in.SessionID + out := &types.RunnerScaleSetSession{ + SessionId: &sessionId, + OwnerName: in.OwnerName, + MessageQueueUrl: in.MessageQueueURL, + MessageQueueAccessToken: in.MessageQueueAccessToken, + Statistics: toStatistics(in.Statistics), + } + if in.RunnerScaleSet != nil { + out.RunnerScaleSet = toScaleSet(in.RunnerScaleSet) + } + return out +} + +func toStatistics(in *scaleset.RunnerScaleSetStatistic) *types.RunnerScaleSetStatistic { + if in == nil { + return nil + } + return &types.RunnerScaleSetStatistic{ + TotalAvailableJobs: in.TotalAvailableJobs, + TotalAcquiredJobs: in.TotalAcquiredJobs, + TotalAssignedJobs: in.TotalAssignedJobs, + TotalRunningJobs: in.TotalRunningJobs, + TotalRegisteredRunners: in.TotalRegisteredRunners, + TotalBusyRunners: in.TotalBusyRunners, + TotalIdleRunners: in.TotalIdleRunners, + } +} diff --git a/pkg/github/scalesetclient/client_test.go b/pkg/github/scalesetclient/client_test.go new file mode 100644 index 0000000..affbfe4 --- /dev/null +++ b/pkg/github/scalesetclient/client_test.go @@ -0,0 +1,210 @@ +package scalesetclient + +import ( + "context" + "net/http" + "testing" + + "github.com/macstadium/orka-github-actions-integration/pkg/env" + "github.com/macstadium/orka-github-actions-integration/pkg/logging" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestScaleSetClient(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "ScaleSet Client Suite") +} + +const maxRunners = 5 + +func newClientAgainst(s *stub) *Client { + client, err := newSDKClientForURL(s.configURL()) + Expect(err).NotTo(HaveOccurred()) + return client +} + +func newSDKClientForURL(configURL string) (*Client, error) { + return New(&env.Data{ + GitHubURL: configURL, + GitHubAppID: 123, + GitHubAppInstallationID: 456, + GitHubAppPrivateKey: mustAppPrivateKey(), + }, maxRunners) +} + +var _ = BeforeSuite(func() { + logging.SetupLogger("error") +}) + +var _ = Describe("SDK-backed client against a GHES-shaped stub", func() { + var ( + s *stub + ctx context.Context + client *Client + ) + + BeforeEach(func() { + s = newStub() + ctx = context.Background() + }) + + AfterEach(func() { + s.close() + }) + + Describe("startup with jobs already in available", func() { + BeforeEach(func() { + s.setStatistics(map[string]int{"totalAvailableJobs": 2, "totalAssignedJobs": 0}) + client = newClientAgainst(s) + }) + + It("reports no acquirable jobs, because the SDK dropped that endpoint", func() { + session, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).NotTo(HaveOccurred()) + Expect(session.Statistics.TotalAvailableJobs).To(Equal(2)) + + jobs, err := client.GetAcquirableJobs(ctx, 1) + Expect(err).NotTo(HaveOccurred()) + Expect(jobs.Jobs).To(BeEmpty(), "the acquirablejobs endpoint is never called") + + _, _, acquired := s.observed() + Expect(acquired).To(BeEmpty()) + }) + + It("recovers the pre-queued jobs when the broker delivers them", func() { + s.messageEnvelope = jobMessagesEnvelope(1, s.statisticsBody(), []map[string]any{ + jobAvailable(1001), jobAvailable(1002), + }) + + _, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).NotTo(HaveOccurred()) + + msg, err := client.GetMessage(ctx, "", "", 0) + Expect(err).NotTo(HaveOccurred()) + Expect(msg).NotTo(BeNil()) + Expect(msg.MessageType).To(Equal(jobMessagesType)) + Expect(msg.Body).To(ContainSubstring("1001")) + Expect(msg.Body).To(ContainSubstring("1002")) + + _, err = client.AcquireJobs(ctx, 1, "", []int64{1001, 1002}) + Expect(err).NotTo(HaveOccurred()) + + _, _, acquired := s.observed() + Expect(acquired).To(Equal([][]int64{{1001, 1002}})) + }) + + It("strands the jobs when the broker stays silent", func() { + s.messageQueueEmpty = true + + session, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).NotTo(HaveOccurred()) + Expect(session.Statistics.TotalAvailableJobs).To(Equal(2)) + + for range 3 { + msg, err := client.GetMessage(ctx, "", "", 0) + Expect(err).NotTo(HaveOccurred()) + Expect(msg).To(BeNil(), "202 Accepted means no message") + } + + calls, _, acquired := s.observed() + Expect(calls).To(Equal(3)) + Expect(acquired).To(BeEmpty(), "nothing was ever acquired despite 2 available jobs") + }) + }) + + Describe("unrecognised outer message type", func() { + BeforeEach(func() { + client = newClientAgainst(s) + }) + + It("returns an error instead of skipping the message", func() { + s.messageEnvelope = typedEnvelope(1, "SomeFutureMessageType", s.statisticsBody(), nil) + + _, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).NotTo(HaveOccurred()) + + _, err = client.GetMessage(ctx, "", "", 0) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("unsupported message type")) + }) + + It("tolerates unrecognised inner job message types", func() { + s.messageEnvelope = jobMessagesEnvelope(1, s.statisticsBody(), []map[string]any{ + {"messageType": "SomeFutureJobMessage", "runnerRequestId": 99}, + }) + + _, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).NotTo(HaveOccurred()) + + msg, err := client.GetMessage(ctx, "", "", 0) + Expect(err).NotTo(HaveOccurred()) + Expect(msg).NotTo(BeNil()) + }) + }) + + Describe("X-ScaleSetMaxCapacity header", func() { + BeforeEach(func() { + client = newClientAgainst(s) + }) + + It("is sent on every GetMessage", func() { + s.messageQueueEmpty = true + + _, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).NotTo(HaveOccurred()) + + _, err = client.GetMessage(ctx, "", "", 0) + Expect(err).NotTo(HaveOccurred()) + + _, headers, _ := s.observed() + Expect(headers).To(ConsistOf("5")) + }) + + It("surfaces an error if the server rejects the header", func() { + s.messageStatus = http.StatusBadRequest + + _, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).NotTo(HaveOccurred()) + + _, err = client.GetMessage(ctx, "", "", 0) + Expect(err).To(HaveOccurred()) + }) + }) + + Describe("409 session conflict", func() { + BeforeEach(func() { + s.sessionStatus = http.StatusConflict + client = newClientAgainst(s) + }) + + It("surfaces the conflict only as text, not as an inspectable status", func() { + _, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("409")) + }) + }) + + Describe("scale set and runner operations", func() { + BeforeEach(func() { + client = newClientAgainst(s) + }) + + It("round-trips a scale set lookup", func() { + scaleSet, err := client.GetRunnerScaleSet(ctx, 1, "orka-runners") + Expect(err).NotTo(HaveOccurred()) + Expect(scaleSet).NotTo(BeNil()) + Expect(scaleSet.Id).To(Equal(1)) + Expect(scaleSet.Name).To(Equal("orka-runners")) + Expect(scaleSet.RunnerSetting.DisableUpdate).To(BeTrue()) + Expect(scaleSet.Labels).To(HaveLen(1)) + }) + + It("generates a JIT runner config", func() { + cfg, err := client.CreateRunner(ctx, 1, "runner-7") + Expect(err).NotTo(HaveOccurred()) + Expect(cfg.EncodedJITConfig).To(Equal("encoded-jit-config")) + Expect(cfg.Runner.Id).To(Equal(7)) + }) + }) +}) diff --git a/pkg/github/scalesetclient/config_test.go b/pkg/github/scalesetclient/config_test.go new file mode 100644 index 0000000..84bbf46 --- /dev/null +++ b/pkg/github/scalesetclient/config_test.go @@ -0,0 +1,29 @@ +package scalesetclient + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("config URL handling", func() { + DescribeTable("accepted config URLs", + func(configURL string) { + _, err := newSDKClientForURL(configURL) + Expect(err).NotTo(HaveOccurred()) + }, + Entry("organization", "https://github.com/testorg"), + Entry("repository", "https://github.com/testorg/testrepo"), + Entry("GHES organization", "https://ghes.example.com/testorg"), + Entry("GHE.com data residency", "https://acme.ghe.com/testorg"), + Entry("enterprise", "https://github.com/enterprises/acme"), + ) + + DescribeTable("rejected config URLs", + func(configURL string) { + _, err := newSDKClientForURL(configURL) + Expect(err).To(HaveOccurred()) + }, + Entry("no path", "https://github.com"), + Entry("too many path segments", "https://github.com/a/b/c"), + ) +}) diff --git a/pkg/github/scalesetclient/stub_test.go b/pkg/github/scalesetclient/stub_test.go new file mode 100644 index 0000000..ad04b3c --- /dev/null +++ b/pkg/github/scalesetclient/stub_test.go @@ -0,0 +1,248 @@ +package scalesetclient + +import ( + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "encoding/json" + "encoding/pem" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "time" + + "github.com/golang-jwt/jwt/v4" + "github.com/google/uuid" +) + +type stub struct { + server *httptest.Server + + mu sync.Mutex + + sessionStatus int + messageStatus int + messageEnvelope any + messageQueueEmpty bool + + maxCapacityHeaders []string + acquiredIDs [][]int64 + getMessageCalls int + + statistics map[string]int +} + +func newStub() *stub { + s := &stub{ + sessionStatus: http.StatusOK, + messageStatus: http.StatusOK, + statistics: map[string]int{ + "totalAvailableJobs": 0, + "totalAssignedJobs": 0, + "totalRegisteredRunners": 0, + "totalRunningJobs": 0, + }, + } + s.server = httptest.NewServer(http.HandlerFunc(s.route)) + return s +} + +func (s *stub) close() { s.server.Close() } + +func (s *stub) url() string { return s.server.URL } + +func (s *stub) configURL() string { return s.server.URL + "/testorg" } + +func (s *stub) setStatistics(stats map[string]int) { + s.mu.Lock() + defer s.mu.Unlock() + for k, v := range stats { + s.statistics[k] = v + } +} + +func (s *stub) observed() (getMessageCalls int, capacityHeaders []string, acquired [][]int64) { + s.mu.Lock() + defer s.mu.Unlock() + return s.getMessageCalls, append([]string(nil), s.maxCapacityHeaders...), append([][]int64(nil), s.acquiredIDs...) +} + +func (s *stub) statisticsBody() map[string]int { + s.mu.Lock() + defer s.mu.Unlock() + out := map[string]int{} + for k, v := range s.statistics { + out[k] = v + } + return out +} + +func (s *stub) route(w http.ResponseWriter, r *http.Request) { + path := r.URL.Path + + switch { + case strings.HasSuffix(path, "/access_tokens"): + writeJSON(w, http.StatusCreated, map[string]any{ + "token": "installation-token", + "expires_at": time.Now().Add(time.Hour), + }) + + case strings.HasSuffix(path, "/actions/runners/registration-token"): + writeJSON(w, http.StatusCreated, map[string]any{ + "token": "registration-token", + "expires_at": time.Now().Add(time.Hour), + }) + + case path == "/api/v3/actions/runner-registration": + writeJSON(w, http.StatusOK, map[string]any{ + "url": s.url(), + "token": mustAdminJWT(), + }) + + case path == "/_apis/runtime/runnerscalesets" && r.Method == http.MethodGet: + writeJSON(w, http.StatusOK, map[string]any{ + "count": 1, + "value": []map[string]any{s.scaleSetBody()}, + }) + + case path == "/_apis/runtime/runnerscalesets" && r.Method == http.MethodPost: + writeJSON(w, http.StatusOK, s.scaleSetBody()) + + case strings.HasSuffix(path, "/acquirejobs"): + var ids []int64 + _ = json.NewDecoder(r.Body).Decode(&ids) + s.mu.Lock() + s.acquiredIDs = append(s.acquiredIDs, ids) + s.mu.Unlock() + writeJSON(w, http.StatusOK, map[string]any{"count": len(ids), "value": ids}) + + case strings.HasSuffix(path, "/generatejitconfig"): + writeJSON(w, http.StatusOK, map[string]any{ + "runner": map[string]any{"id": 7, "name": "runner-7", "runnerScaleSetId": 1}, + "encodedJITConfig": "encoded-jit-config", + }) + + case strings.Contains(path, "/sessions") && r.Method == http.MethodPost: + s.mu.Lock() + status := s.sessionStatus + s.mu.Unlock() + + if status != http.StatusOK { + writeJSON(w, status, map[string]any{ + "typeName": "SessionConflictException", + "message": "runner scale set already has an active session", + }) + return + } + + writeJSON(w, http.StatusOK, map[string]any{ + "sessionId": uuid.New().String(), + "ownerName": "stub-owner", + "messageQueueUrl": s.url() + "/messages", + "messageQueueAccessToken": mustAdminJWT(), + "runnerScaleSet": s.scaleSetBody(), + "statistics": s.statisticsBody(), + }) + + case strings.Contains(path, "/sessions") && r.Method == http.MethodDelete: + w.WriteHeader(http.StatusNoContent) + + case path == "/messages" && r.Method == http.MethodGet: + s.mu.Lock() + s.getMessageCalls++ + s.maxCapacityHeaders = append(s.maxCapacityHeaders, r.Header.Get("X-ScaleSetMaxCapacity")) + status, envelope, empty := s.messageStatus, s.messageEnvelope, s.messageQueueEmpty + s.mu.Unlock() + + if empty || status == http.StatusAccepted { + w.WriteHeader(http.StatusAccepted) + return + } + if status != http.StatusOK { + writeJSON(w, status, map[string]any{"message": "stub failure"}) + return + } + writeJSON(w, http.StatusOK, envelope) + + case strings.HasPrefix(path, "/messages/") && r.Method == http.MethodDelete: + w.WriteHeader(http.StatusNoContent) + + default: + http.Error(w, fmt.Sprintf("stub: unhandled %s %s", r.Method, path), http.StatusNotFound) + } +} + +func (s *stub) scaleSetBody() map[string]any { + return map[string]any{ + "id": 1, + "name": "orka-runners", + "runnerGroupId": 1, + "labels": []map[string]any{{"name": "orka-runners", "type": "System"}}, + "RunnerSetting": map[string]any{"disableUpdate": true}, + "statistics": s.statisticsBody(), + } +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + if body != nil { + _ = json.NewEncoder(w).Encode(body) + } +} + +func jobMessagesEnvelope(messageID int, statistics map[string]int, jobs []map[string]any) map[string]any { + return typedEnvelope(messageID, "RunnerScaleSetJobMessages", statistics, jobs) +} + +func typedEnvelope(messageID int, messageType string, statistics map[string]int, jobs []map[string]any) map[string]any { + body := "" + if len(jobs) > 0 { + encoded, err := json.Marshal(jobs) + if err != nil { + panic(err) + } + body = string(encoded) + } + + return map[string]any{ + "messageId": messageID, + "messageType": messageType, + "body": body, + "statistics": statistics, + } +} + +func jobAvailable(runnerRequestID int64) map[string]any { + return map[string]any{ + "messageType": "JobAvailable", + "runnerRequestId": runnerRequestID, + "jobId": fmt.Sprintf("job-%d", runnerRequestID), + "acquireJobUrl": "http://stub/acquire", + } +} + +func mustAdminJWT() string { + token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.RegisteredClaims{ + ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour)), + IssuedAt: jwt.NewNumericDate(time.Now()), + }) + signed, err := token.SignedString([]byte("stub-secret")) + if err != nil { + panic(err) + } + return signed +} + +func mustAppPrivateKey() string { + key, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + panic(err) + } + return string(pem.EncodeToMemory(&pem.Block{ + Type: "RSA PRIVATE KEY", + Bytes: x509.MarshalPKCS1PrivateKey(key), + })) +} diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 686cac7..31d732d 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -84,7 +84,7 @@ func Start( ctx context.Context, logger *zap.SugaredLogger, envData *env.Data, - actionsClient *actions.ActionsClient, + actionsClient actions.ActionsService, runnerName string, groupId int, ) *Metrics { @@ -122,7 +122,7 @@ func (m *Metrics) startPoller( ctx context.Context, logger *zap.SugaredLogger, interval time.Duration, - actionsClient *actions.ActionsClient, + actionsClient actions.ActionsService, runnerName string, groupId int, ) { From fa53cace6375cab25865b4204608ed9842c69eed Mon Sep 17 00:00:00 2001 From: Roy Larsen Date: Tue, 4 Aug 2026 12:10:47 -0400 Subject: [PATCH 2/5] Set MAX_RUNNERS to 9000 --- pkg/env/env.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/env/env.go b/pkg/env/env.go index e045290..312de83 100644 --- a/pkg/env/env.go +++ b/pkg/env/env.go @@ -91,7 +91,7 @@ func ParseEnv() *Data { VMTrackerInterval: getDurationEnv(VMTrackerIntervalEnvName, 300*time.Second), - MaxRunners: getIntEnv(MaxRunnersEnvName, 10), + MaxRunners: getIntEnv(MaxRunnersEnvName, 9000), LogLevel: getEnvWithDefault(LogLevelEnvName, logging.LogLevelInfo), From 31517ddec79aa63467ab194478fea1c0b23e4f62 Mon Sep 17 00:00:00 2001 From: Roy Larsen Date: Tue, 4 Aug 2026 12:11:35 -0400 Subject: [PATCH 3/5] Detect skippable message --- pkg/github/scalesetclient/client.go | 97 +++++++++++++++++++++++- pkg/github/scalesetclient/client_test.go | 23 ++++-- 2 files changed, 110 insertions(+), 10 deletions(-) diff --git a/pkg/github/scalesetclient/client.go b/pkg/github/scalesetclient/client.go index 4ba4484..74db90a 100644 --- a/pkg/github/scalesetclient/client.go +++ b/pkg/github/scalesetclient/client.go @@ -5,17 +5,29 @@ import ( "encoding/json" "fmt" "log/slog" + "net/http" + "net/url" + "regexp" + "strconv" + "strings" "sync" "github.com/actions/scaleset" "github.com/google/uuid" "github.com/macstadium/orka-github-actions-integration/pkg/env" + "github.com/macstadium/orka-github-actions-integration/pkg/github/actions" "github.com/macstadium/orka-github-actions-integration/pkg/github/types" "github.com/macstadium/orka-github-actions-integration/pkg/logging" "go.uber.org/zap/exp/zapslog" ) -const jobMessagesType = "RunnerScaleSetJobMessages" +const ( + jobMessagesType = "RunnerScaleSetJobMessages" + messageAPIVersion = "application/json; api-version=6.0-preview" + unsupportedTypeText = "unsupported message type" +) + +var statusCodePattern = regexp.MustCompile(`status="(\d{3})`) type Client struct { sdk *scaleset.Client @@ -96,7 +108,7 @@ func (c *Client) DeleteRunner(ctx context.Context, runnerID int) error { func (c *Client) CreateMessageSession(ctx context.Context, runnerScaleSetId int, owner string) (*types.RunnerScaleSetSession, error) { sessionClient, err := c.sdk.MessageSessionClient(ctx, runnerScaleSetId, owner) if err != nil { - return nil, err + return nil, withStatusCode(err) } c.mu.Lock() @@ -137,9 +149,15 @@ func (c *Client) GetMessage(ctx context.Context, messageQueueUrl, messageQueueAc } msg, err := session.GetMessage(ctx, int(lastMessageId), c.maxRunners) - if err != nil || msg == nil { + if err != nil { + if strings.Contains(err.Error(), unsupportedTypeText) { + return c.skippableMessage(ctx, session, lastMessageId) + } return nil, err } + if msg == nil { + return nil, nil + } body, err := encodeBody(msg) if err != nil { @@ -166,6 +184,79 @@ func (c *Client) GetAcquirableJobs(ctx context.Context, runnerScaleSetId int) (* return &types.AcquirableJobList{Count: 0, Jobs: []types.AcquirableJob{}}, nil } +func (c *Client) skippableMessage(ctx context.Context, session *scaleset.MessageSessionClient, lastMessageId int64) (*types.RunnerScaleSetMessage, error) { + current := session.Session() + + target, err := url.Parse(current.MessageQueueURL) + if err != nil { + return nil, fmt.Errorf("failed to parse message queue url: %w", err) + } + if lastMessageId > 0 { + query := target.Query() + query.Set("lastMessageId", strconv.FormatInt(lastMessageId, 10)) + target.RawQuery = query.Encode() + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil) + if err != nil { + return nil, err + } + req.Header.Set("Accept", messageAPIVersion) + req.Header.Set("Authorization", "Bearer "+current.MessageQueueAccessToken) + req.Header.Set(scaleset.HeaderScaleSetMaxCapacity, strconv.Itoa(c.maxRunners)) + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("failed to re-read unsupported message: status %s", resp.Status) + } + + var envelope struct { + MessageID int64 `json:"messageId"` + MessageType string `json:"messageType"` + Statistics *types.RunnerScaleSetStatistic `json:"statistics"` + } + if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil { + return nil, fmt.Errorf("failed to decode unsupported message envelope: %w", err) + } + + if envelope.MessageType == jobMessagesType { + return nil, fmt.Errorf("message %d is decodable on retry, deferring to the SDK", envelope.MessageID) + } + + c.logger.Warn("re-read unsupported message type so it can be acknowledged", + slog.Int64("messageId", envelope.MessageID), + slog.String("messageType", envelope.MessageType)) + + return &types.RunnerScaleSetMessage{ + MessageId: envelope.MessageID, + MessageType: envelope.MessageType, + Statistics: envelope.Statistics, + }, nil +} + +func withStatusCode(err error) error { + if err == nil { + return nil + } + + match := statusCodePattern.FindStringSubmatch(err.Error()) + if match == nil { + return err + } + + status, convErr := strconv.Atoi(match[1]) + if convErr != nil { + return err + } + + return &actions.ActionsError{StatusCode: status, Message: err.Error()} +} + func (c *Client) currentSession() (*scaleset.MessageSessionClient, error) { c.mu.Lock() defer c.mu.Unlock() diff --git a/pkg/github/scalesetclient/client_test.go b/pkg/github/scalesetclient/client_test.go index affbfe4..592077d 100644 --- a/pkg/github/scalesetclient/client_test.go +++ b/pkg/github/scalesetclient/client_test.go @@ -2,10 +2,12 @@ package scalesetclient import ( "context" + "errors" "net/http" "testing" "github.com/macstadium/orka-github-actions-integration/pkg/env" + "github.com/macstadium/orka-github-actions-integration/pkg/github/actions" "github.com/macstadium/orka-github-actions-integration/pkg/logging" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -118,15 +120,19 @@ var _ = Describe("SDK-backed client against a GHES-shaped stub", func() { client = newClientAgainst(s) }) - It("returns an error instead of skipping the message", func() { - s.messageEnvelope = typedEnvelope(1, "SomeFutureMessageType", s.statisticsBody(), nil) + It("returns a skippable message so it can be acknowledged and deleted", func() { + s.messageEnvelope = typedEnvelope(41, "SomeFutureMessageType", s.statisticsBody(), nil) _, err := client.CreateMessageSession(ctx, 1, "owner") Expect(err).NotTo(HaveOccurred()) - _, err = client.GetMessage(ctx, "", "", 0) - Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(ContainSubstring("unsupported message type")) + msg, err := client.GetMessage(ctx, "", "", 0) + Expect(err).NotTo(HaveOccurred()) + Expect(msg).NotTo(BeNil()) + Expect(msg.MessageId).To(Equal(int64(41))) + Expect(msg.MessageType).To(Equal("SomeFutureMessageType")) + Expect(msg.Body).To(BeEmpty()) + Expect(msg.Statistics).NotTo(BeNil(), "the processor rejects messages with nil statistics") }) It("tolerates unrecognised inner job message types", func() { @@ -178,10 +184,13 @@ var _ = Describe("SDK-backed client against a GHES-shaped stub", func() { client = newClientAgainst(s) }) - It("surfaces the conflict only as text, not as an inspectable status", func() { + It("surfaces the conflict as an inspectable ActionsError so recovery still fires", func() { _, err := client.CreateMessageSession(ctx, 1, "owner") Expect(err).To(HaveOccurred()) - Expect(err.Error()).To(ContainSubstring("409")) + + actionsErr := &actions.ActionsError{} + Expect(errors.As(err, &actionsErr)).To(BeTrue()) + Expect(actionsErr.StatusCode).To(Equal(http.StatusConflict)) }) }) From ba928831b42893e708b8e4ce8f468266b87decdf Mon Sep 17 00:00:00 2001 From: Roy Larsen Date: Tue, 4 Aug 2026 12:22:27 -0400 Subject: [PATCH 4/5] Warn on GITHUB_API_URL set --- main.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/main.go b/main.go index e091e6f..0921813 100644 --- a/main.go +++ b/main.go @@ -33,6 +33,10 @@ func main() { logging.SetupLogger(envData.LogLevel) logger := logging.Logger.Named("main") + if apiUrl := os.Getenv(env.GitHubAPIURLEnvName); apiUrl != "" { + logger.Warnf("%s is set to %q but is no longer used. The GitHub API URL is derived from %s; the derived value is logged by the scaleset client below. Remove %s to avoid confusion.", env.GitHubAPIURLEnvName, apiUrl, env.GitHubURLEnvName, env.GitHubAPIURLEnvName) + } + runnerName := envData.Runners[0].Name groupId := constants.DefaultRunnerGroupID if envData.Runners[0].Id != 0 { From 37469000d1c5c0d334d9a394898bac5cc7716b27 Mon Sep 17 00:00:00 2001 From: Roy Larsen Date: Thu, 20 Aug 2026 10:51:59 -0400 Subject: [PATCH 5/5] Handle 404 to recreate if expected --- main.go | 13 +++++++++---- pkg/github/runners/manager.go | 8 +++++++- pkg/github/scalesetclient/client_test.go | 16 ++++++++++++++++ 3 files changed, 32 insertions(+), 5 deletions(-) diff --git a/main.go b/main.go index 0921813..3b1bb51 100644 --- a/main.go +++ b/main.go @@ -98,11 +98,16 @@ func main() { } runnerManager, err := runners.NewRunnerManager(ctx, actionsClient, runnerScaleSet.Id) - if errors.Is(err, runners.ErrActiveSession) { - logger.Infof("scale set %s (id=%d) has a stale active session, deleting and recreating", runnerScaleSet.Name, runnerScaleSet.Id) - if err = actionsClient.DeleteRunnerScaleSet(ctx, runnerScaleSet.Id); err != nil { - panic(fmt.Sprintf("error deleting scale set with active session: %s", err.Error())) + if errors.Is(err, runners.ErrActiveSession) || errors.Is(err, runners.ErrScaleSetNotFound) { + if errors.Is(err, runners.ErrActiveSession) { + logger.Infof("scale set %s (id=%d) has a stale active session, deleting and recreating", runnerScaleSet.Name, runnerScaleSet.Id) + if err = actionsClient.DeleteRunnerScaleSet(ctx, runnerScaleSet.Id); err != nil { + panic(fmt.Sprintf("error deleting scale set with active session: %s", err.Error())) + } + } else { + logger.Infof("scale set %s (id=%d) was listed but no longer exists, recreating", runnerScaleSet.Name, runnerScaleSet.Id) } + runnerScaleSet, err = createScaleSet(ctx, actionsClient, runnerName, groupId) if err != nil { panic(fmt.Sprintf("error recreating scale set after active session conflict: %s", err.Error())) diff --git a/pkg/github/runners/manager.go b/pkg/github/runners/manager.go index 7b20078..ab51d09 100644 --- a/pkg/github/runners/manager.go +++ b/pkg/github/runners/manager.go @@ -27,7 +27,10 @@ const ( runnerScaleSetJobMessagesType = "RunnerScaleSetJobMessages" ) -var ErrActiveSession = errors.New("runner scale set already has an active session") +var ( + ErrActiveSession = errors.New("runner scale set already has an active session") + ErrScaleSetNotFound = errors.New("runner scale set no longer exists") +) func NewRunnerManager(ctx context.Context, client actions.ActionsService, runnerScaleSetId int) (*RunnerManager, error) { logger := logging.Logger.Named(fmt.Sprintf("runner-manager-%d", runnerScaleSetId)) @@ -75,6 +78,9 @@ func createSessionWithRetry(ctx context.Context, logger *zap.SugaredLogger, clie if actionsErr.StatusCode == http.StatusConflict { return nil, fmt.Errorf("%w: %s", ErrActiveSession, err) } + if actionsErr.StatusCode == http.StatusNotFound { + return nil, fmt.Errorf("%w: %s", ErrScaleSetNotFound, err) + } logger.Infof("unable to create message session, client-side error (status %d), won't retry: %s", actionsErr.StatusCode, err.Error()) return nil, fmt.Errorf("create message session http request failed. %w", err) } diff --git a/pkg/github/scalesetclient/client_test.go b/pkg/github/scalesetclient/client_test.go index 592077d..1f1619b 100644 --- a/pkg/github/scalesetclient/client_test.go +++ b/pkg/github/scalesetclient/client_test.go @@ -194,6 +194,22 @@ var _ = Describe("SDK-backed client against a GHES-shaped stub", func() { }) }) + Describe("404 scale set not found", func() { + BeforeEach(func() { + s.sessionStatus = http.StatusNotFound + client = newClientAgainst(s) + }) + + It("surfaces the missing scale set as an inspectable ActionsError so recovery still fires", func() { + _, err := client.CreateMessageSession(ctx, 1, "owner") + Expect(err).To(HaveOccurred()) + + actionsErr := &actions.ActionsError{} + Expect(errors.As(err, &actionsErr)).To(BeTrue()) + Expect(actionsErr.StatusCode).To(Equal(http.StatusNotFound)) + }) + }) + Describe("scale set and runner operations", func() { BeforeEach(func() { client = newClientAgainst(s)