fix: 等待 Runner facade 就绪
This commit is contained in:
@@ -79,6 +79,19 @@ func RunExecutor(ctx context.Context, config ExecutorConfig) error {
|
|||||||
server := &http.Server{Handler: proxy.Handler, ReadHeaderTimeout: 10 * time.Second}
|
server := &http.Server{Handler: proxy.Handler, ReadHeaderTimeout: 10 * time.Second}
|
||||||
serverErrors := make(chan error, 1)
|
serverErrors := make(chan error, 1)
|
||||||
go func() { serverErrors <- server.Serve(listener) }()
|
go func() { serverErrors <- server.Serve(listener) }()
|
||||||
|
readyContext, cancelReady := context.WithTimeout(ctx, 2*time.Minute)
|
||||||
|
readyErr := waitForFacade(readyContext, "http://"+listener.Addr().String())
|
||||||
|
cancelReady()
|
||||||
|
if readyErr != nil {
|
||||||
|
shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||||
|
shutdownErr := server.Shutdown(shutdownContext)
|
||||||
|
cancel()
|
||||||
|
serverErr := <-serverErrors
|
||||||
|
if errors.Is(serverErr, http.ErrServerClosed) {
|
||||||
|
serverErr = nil
|
||||||
|
}
|
||||||
|
return errors.Join(readyErr, shutdownErr, serverErr)
|
||||||
|
}
|
||||||
|
|
||||||
command := exec.CommandContext(ctx, config.RunnerBinary, "daemon", "--once")
|
command := exec.CommandContext(ctx, config.RunnerBinary, "daemon", "--once")
|
||||||
command.Dir = workDir
|
command.Dir = workDir
|
||||||
@@ -95,6 +108,30 @@ func RunExecutor(ctx context.Context, config ExecutorConfig) error {
|
|||||||
return errors.Join(runnerErr, shutdownErr, serverErr)
|
return errors.Join(runnerErr, shutdownErr, serverErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func waitForFacade(ctx context.Context, endpoint string) error {
|
||||||
|
client := &http.Client{Timeout: 2 * time.Second}
|
||||||
|
ticker := time.NewTicker(250 * time.Millisecond)
|
||||||
|
defer ticker.Stop()
|
||||||
|
for {
|
||||||
|
request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("create facade readiness request: %w", err)
|
||||||
|
}
|
||||||
|
response, requestErr := client.Do(request)
|
||||||
|
if requestErr == nil {
|
||||||
|
_ = response.Body.Close()
|
||||||
|
if response.StatusCode < http.StatusInternalServerError {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return fmt.Errorf("wait for runner facade: %w", ctx.Err())
|
||||||
|
case <-ticker.C:
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// ExecutorConfigFromEnvironment reads the non-secret image configuration and
|
// ExecutorConfigFromEnvironment reads the non-secret image configuration and
|
||||||
// the assignment-scoped values injected by the backend. The Workload API
|
// the assignment-scoped values injected by the backend. The Workload API
|
||||||
// address follows SPIFFE_ENDPOINT_SOCKET through go-spiffe when not set here.
|
// address follows SPIFFE_ENDPOINT_SOCKET through go-spiffe when not set here.
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
package runnerbootstrap
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestWaitForFacadeRetriesTransientGatewayFailure(t *testing.T) {
|
||||||
|
var requests atomic.Int32
|
||||||
|
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
|
||||||
|
if requests.Add(1) < 3 {
|
||||||
|
writer.WriteHeader(http.StatusBadGateway)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writer.WriteHeader(http.StatusNotFound)
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
if err := waitForFacade(ctx, server.URL); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if requests.Load() != 3 {
|
||||||
|
t.Fatalf("requests = %d", requests.Load())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestWaitForFacadeStopsWithContext(t *testing.T) {
|
||||||
|
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
|
||||||
|
writer.WriteHeader(http.StatusServiceUnavailable)
|
||||||
|
}))
|
||||||
|
defer server.Close()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
|
||||||
|
defer cancel()
|
||||||
|
if err := waitForFacade(ctx, server.URL); err == nil {
|
||||||
|
t.Fatal("expected readiness timeout")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user