feat: run gateway runtime reporter loop
This commit is contained in:
parent
5a1873a9f0
commit
de2c9ce9b6
@ -134,10 +134,40 @@ func (reporter *RuntimeReporter) Report(ctx context.Context) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Run keeps the Controller session alive through bounded, periodic Runtime
|
||||||
|
// reports. The snapshot stream may start after Register, so an absent local
|
||||||
|
// snapshot is a normal waiting state rather than a process failure.
|
||||||
|
func (reporter *RuntimeReporter) Run(ctx context.Context) error {
|
||||||
|
registration, err := reporter.Register(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
ticker := time.NewTicker(registration.HeartbeatInterval)
|
||||||
|
defer ticker.Stop()
|
||||||
|
for {
|
||||||
|
if err := reporter.reportIfSnapshot(ctx); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return ctx.Err()
|
||||||
|
case <-ticker.C:
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (reporter *RuntimeReporter) reportIfSnapshot(ctx context.Context) error {
|
||||||
|
err := reporter.Report(ctx)
|
||||||
|
if errors.Is(err, snapshot.ErrInvalidRuntimeReport) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
func validateRegistration(workerID string, response *controlplanev1.RegisterWorkerResponse) (Registration, error) {
|
func validateRegistration(workerID string, response *controlplanev1.RegisterWorkerResponse) (Registration, error) {
|
||||||
if response == nil || response.GetWorkerId() != workerID || !workerruntime.ValidIdentifier(response.GetSessionId()) ||
|
if response == nil || response.GetWorkerId() != workerID || !workerruntime.ValidIdentifier(response.GetSessionId()) ||
|
||||||
response.GetOwnershipEpoch() == 0 || response.GetHeartbeatInterval() == nil || response.GetMaxStaleAge() == nil ||
|
response.GetOwnershipEpoch() == 0 || response.GetHeartbeatInterval() == nil || response.GetMaxStaleAge() == nil ||
|
||||||
response.GetHeartbeatInterval().AsDuration() <= 0 || response.GetMaxStaleAge().AsDuration() <= 0 {
|
response.GetHeartbeatInterval().AsDuration() <= 0 || response.GetMaxStaleAge().AsDuration() < response.GetHeartbeatInterval().AsDuration() {
|
||||||
return Registration{}, ErrInvalidOptions
|
return Registration{}, ErrInvalidOptions
|
||||||
}
|
}
|
||||||
return Registration{
|
return Registration{
|
||||||
|
|||||||
@ -88,6 +88,39 @@ func TestRuntimeReporterRejectsInvalidState(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestRuntimeReporterRunWaitsForSnapshotThenReports(t *testing.T) {
|
||||||
|
store := snapshot.NewStore("cluster-a", "worker-a")
|
||||||
|
client := &runClient{runtime: make(chan *controlplanev1.ReportRuntimeRequest, 8)}
|
||||||
|
reporter, err := NewRuntimeReporter(client, store, Options{
|
||||||
|
WorkerID: "worker-a", InstanceID: "instance-a", Zone: "zone-a", ProtocolVersion: 1, Now: time.Now,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("NewRuntimeReporter(): %v", err)
|
||||||
|
}
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
result := make(chan error, 1)
|
||||||
|
go func() { result <- reporter.Run(ctx) }()
|
||||||
|
time.Sleep(15 * time.Millisecond)
|
||||||
|
select {
|
||||||
|
case request := <-client.runtime:
|
||||||
|
t.Fatalf("ReportRuntime() before snapshot = %+v", request)
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
applySnapshot(t, store)
|
||||||
|
select {
|
||||||
|
case request := <-client.runtime:
|
||||||
|
if request.GetReportSequence() != 1 {
|
||||||
|
t.Fatalf("report sequence = %d, want 1", request.GetReportSequence())
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("ReportRuntime() was not called after applying a snapshot")
|
||||||
|
}
|
||||||
|
cancel()
|
||||||
|
if err := <-result; !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("Run() error = %v, want context canceled", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
type clientStub struct {
|
type clientStub struct {
|
||||||
register *controlplanev1.RegisterWorkerRequest
|
register *controlplanev1.RegisterWorkerRequest
|
||||||
registration *controlplanev1.RegisterWorkerResponse
|
registration *controlplanev1.RegisterWorkerResponse
|
||||||
@ -118,3 +151,19 @@ func applySnapshot(t *testing.T, store *snapshot.Store) {
|
|||||||
t.Fatalf("Apply(): %v", err)
|
t.Fatalf("Apply(): %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type runClient struct {
|
||||||
|
runtime chan *controlplanev1.ReportRuntimeRequest
|
||||||
|
}
|
||||||
|
|
||||||
|
func (*runClient) RegisterWorker(context.Context, *controlplanev1.RegisterWorkerRequest) (*controlplanev1.RegisterWorkerResponse, error) {
|
||||||
|
return &controlplanev1.RegisterWorkerResponse{
|
||||||
|
WorkerId: "worker-a", SessionId: "session-a", OwnershipEpoch: 7,
|
||||||
|
HeartbeatInterval: durationpb.New(10 * time.Millisecond), MaxStaleAge: durationpb.New(time.Second),
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (client *runClient) ReportRuntime(_ context.Context, request *controlplanev1.ReportRuntimeRequest) (*controlplanev1.ReportRuntimeResponse, error) {
|
||||||
|
client.runtime <- request
|
||||||
|
return &controlplanev1.ReportRuntimeResponse{AcceptedOwnershipEpoch: 7}, nil
|
||||||
|
}
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user