Remove acquire jobs (#4)
This commit is contained in:
Francesco Renzi
@@ -709,97 +709,6 @@ func (c *Client) doSessionRequest(ctx context.Context, method, path string, requ
|
||||
})
|
||||
}
|
||||
|
||||
func (c *Client) AcquireJobs(ctx context.Context, runnerScaleSetID int, messageQueueAccessToken string, requestIDs []int64) ([]int64, error) {
|
||||
u := fmt.Sprintf("%s/%s/%d/acquirejobs?api-version=6.0-preview", c.actionsServiceURL, scaleSetEndpoint, runnerScaleSetID)
|
||||
|
||||
body, err := json.Marshal(requestIDs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal request ids: %w", err)
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewBuffer(body))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create new request with context: %w", err)
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", messageQueueAccessToken))
|
||||
req.Header.Set("User-Agent", c.userAgent.Load())
|
||||
|
||||
resp, err := c.do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to issue the request: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode == http.StatusOK {
|
||||
var acquiredJobs *Int64List
|
||||
if err := json.NewDecoder(resp.Body).Decode(&acquiredJobs); err != nil {
|
||||
return nil, &ActionsError{
|
||||
ActivityID: resp.Header.Get(headerActionsActivityID),
|
||||
StatusCode: resp.StatusCode,
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
|
||||
return acquiredJobs.Value, nil
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusUnauthorized {
|
||||
return nil, parseActionsErrorFromResponse(resp)
|
||||
}
|
||||
|
||||
body, err = io.ReadAll(resp.Body)
|
||||
body = trimByteOrderMark(body)
|
||||
if err != nil {
|
||||
return nil, &ActionsError{
|
||||
ActivityID: resp.Header.Get(headerActionsActivityID),
|
||||
StatusCode: resp.StatusCode,
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
|
||||
return nil, &MessageQueueTokenExpiredError{
|
||||
activityID: resp.Header.Get(headerActionsActivityID),
|
||||
statusCode: resp.StatusCode,
|
||||
msg: string(body),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) GetAcquirableJobs(ctx context.Context, runnerScaleSetID int) (*AcquirableJobList, error) {
|
||||
path := fmt.Sprintf("/%s/%d/acquirablejobs", scaleSetEndpoint, runnerScaleSetID)
|
||||
|
||||
req, err := c.newActionsServiceRequest(ctx, http.MethodGet, path, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create new actions service request: %w", err)
|
||||
}
|
||||
|
||||
resp, err := c.do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to issue the request: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode == http.StatusNoContent {
|
||||
return &AcquirableJobList{Count: 0, Jobs: []AcquirableJob{}}, nil
|
||||
}
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, parseActionsErrorFromResponse(resp)
|
||||
}
|
||||
|
||||
var acquirableJobList *AcquirableJobList
|
||||
if err := json.NewDecoder(resp.Body).Decode(&acquirableJobList); err != nil {
|
||||
return nil, &ActionsError{
|
||||
StatusCode: resp.StatusCode,
|
||||
ActivityID: resp.Header.Get(headerActionsActivityID),
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
|
||||
return acquirableJobList, nil
|
||||
}
|
||||
|
||||
func (c *Client) GenerateJitRunnerConfig(ctx context.Context, jitRunnerSetting *RunnerScaleSetJitRunnerSetting, scaleSetID int) (*RunnerScaleSetJitRunnerConfig, error) {
|
||||
path := fmt.Sprintf("/%s/%d/generatejitconfig", scaleSetEndpoint, scaleSetID)
|
||||
|
||||
|
||||
@@ -1,170 +0,0 @@
|
||||
package scaleset
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestAcquireJobs(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
auth := &ActionsAuth{
|
||||
Token: "token",
|
||||
}
|
||||
|
||||
t.Run("Acquire Job", func(t *testing.T) {
|
||||
want := []int64{1}
|
||||
response := []byte(`{"value": [1]}`)
|
||||
|
||||
session := &RunnerScaleSetSession{
|
||||
RunnerScaleSet: &RunnerScaleSet{ID: 1},
|
||||
MessageQueueAccessToken: "abc",
|
||||
}
|
||||
requestIDs := want
|
||||
|
||||
server := newActionsServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if strings.HasSuffix(r.URL.Path, "/acquirablejobs") {
|
||||
w.Write([]byte(`{"count": 1}`))
|
||||
return
|
||||
}
|
||||
|
||||
w.Write(response)
|
||||
}))
|
||||
|
||||
client, err := NewClient(server.configURLForOrg("my-org"), auth)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = client.GetAcquirableJobs(ctx, 1)
|
||||
require.NoError(t, err)
|
||||
|
||||
got, err := client.AcquireJobs(ctx, session.RunnerScaleSet.ID, session.MessageQueueAccessToken, requestIDs)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, want, got)
|
||||
})
|
||||
|
||||
t.Run("Default retries on server error", func(t *testing.T) {
|
||||
session := &RunnerScaleSetSession{
|
||||
RunnerScaleSet: &RunnerScaleSet{ID: 1},
|
||||
MessageQueueAccessToken: "abc",
|
||||
}
|
||||
requestIDs := []int64{1}
|
||||
|
||||
retryMax := 1
|
||||
actualRetry := 0
|
||||
expectedRetry := retryMax + 1
|
||||
|
||||
server := newActionsServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if strings.HasSuffix(r.URL.Path, "/acquirablejobs") {
|
||||
w.Write([]byte(`{"count": 1}`))
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
actualRetry++
|
||||
}))
|
||||
|
||||
client, err := NewClient(
|
||||
server.configURLForOrg("my-org"),
|
||||
auth,
|
||||
WithRetryMax(retryMax),
|
||||
WithRetryWaitMax(1*time.Millisecond),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = client.GetAcquirableJobs(ctx, 1)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = client.AcquireJobs(context.Background(), session.RunnerScaleSet.ID, session.MessageQueueAccessToken, requestIDs)
|
||||
assert.NotNil(t, err)
|
||||
assert.Equalf(t, actualRetry, expectedRetry, "A retry was expected after the first request but got: %v", actualRetry)
|
||||
})
|
||||
|
||||
t.Run("Should return MessageQueueTokenExpiredError when http error is not Unauthorized", func(t *testing.T) {
|
||||
want := []int64{1}
|
||||
|
||||
session := &RunnerScaleSetSession{
|
||||
RunnerScaleSet: &RunnerScaleSet{ID: 1},
|
||||
MessageQueueAccessToken: "abc",
|
||||
}
|
||||
requestIDs := want
|
||||
|
||||
server := newActionsServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if strings.HasSuffix(r.URL.Path, "/acquirablejobs") {
|
||||
w.Write([]byte(`{"count": 1}`))
|
||||
return
|
||||
}
|
||||
if r.Method == http.MethodPost {
|
||||
http.Error(w, "Session expired", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
}))
|
||||
|
||||
client, err := NewClient(server.configURLForOrg("my-org"), auth)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = client.GetAcquirableJobs(ctx, 1)
|
||||
require.NoError(t, err)
|
||||
|
||||
got, err := client.AcquireJobs(ctx, session.RunnerScaleSet.ID, session.MessageQueueAccessToken, requestIDs)
|
||||
require.Error(t, err)
|
||||
assert.Nil(t, got)
|
||||
var expectedErr *MessageQueueTokenExpiredError
|
||||
assert.True(t, errors.As(err, &expectedErr))
|
||||
})
|
||||
}
|
||||
|
||||
func TestGetAcquirableJobs(t *testing.T) {
|
||||
auth := &ActionsAuth{
|
||||
Token: "token",
|
||||
}
|
||||
|
||||
t.Run("Acquire Job", func(t *testing.T) {
|
||||
want := &AcquirableJobList{}
|
||||
response := []byte(`{"count": 0}`)
|
||||
|
||||
runnerScaleSet := &RunnerScaleSet{ID: 1}
|
||||
|
||||
server := newActionsServer(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Write(response)
|
||||
}))
|
||||
|
||||
client, err := NewClient(server.configURLForOrg("my-org"), auth)
|
||||
require.NoError(t, err)
|
||||
|
||||
got, err := client.GetAcquirableJobs(context.Background(), runnerScaleSet.ID)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, want, got)
|
||||
})
|
||||
|
||||
t.Run("Default retries on server error", func(t *testing.T) {
|
||||
runnerScaleSet := &RunnerScaleSet{ID: 1}
|
||||
|
||||
retryMax := 1
|
||||
|
||||
actualRetry := 0
|
||||
expectedRetry := retryMax + 1
|
||||
|
||||
server := newActionsServer(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
actualRetry++
|
||||
}))
|
||||
|
||||
client, err := NewClient(
|
||||
server.configURLForOrg("my-org"),
|
||||
auth,
|
||||
WithRetryMax(retryMax),
|
||||
WithRetryWaitMax(1*time.Millisecond),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = client.GetAcquirableJobs(context.Background(), runnerScaleSet.ID)
|
||||
require.Error(t, err)
|
||||
assert.Equalf(t, actualRetry, expectedRetry, "A retry was expected after the first request but got: %v", actualRetry)
|
||||
})
|
||||
}
|
||||
@@ -6,22 +6,6 @@ import (
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type AcquirableJobList struct {
|
||||
Count int `json:"count"`
|
||||
Jobs []AcquirableJob `json:"value"`
|
||||
}
|
||||
|
||||
type AcquirableJob struct {
|
||||
AcquireJobURL string `json:"acquireJobUrl"`
|
||||
MessageType string `json:"messageType"`
|
||||
RunnerRequestID int64 `json:"runnerRequestId"`
|
||||
RepositoryName string `json:"repositoryName"`
|
||||
OwnerName string `json:"ownerName"`
|
||||
JobWorkflowRef string `json:"jobWorkflowRef"`
|
||||
EventName string `json:"eventName"`
|
||||
RequestLabels []string `json:"requestLabels"`
|
||||
}
|
||||
|
||||
type Int64List struct {
|
||||
Count int `json:"count"`
|
||||
Value []int64 `json:"value"`
|
||||
|
||||
Reference in New Issue
Block a user