From 8a302cd8cc769f7260a8e2d87664dccbbb86af68 Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Wed, 29 Jul 2026 10:55:19 +0700 Subject: [PATCH] feat(gateway): report provider lifecycle --- cmd/verse-gateway/main.go | 2 +- gateway/control_plane.go | 9 +++++ gateway/gateway_test.go | 28 +++++++++++++-- gateway/provider.go | 4 +-- gateway/transport.go | 71 ++++++++++++++++++++++++++++++++------- go.mod | 2 +- go.sum | 4 +-- 7 files changed, 99 insertions(+), 21 deletions(-) diff --git a/cmd/verse-gateway/main.go b/cmd/verse-gateway/main.go index 951f838..0c3580b 100644 --- a/cmd/verse-gateway/main.go +++ b/cmd/verse-gateway/main.go @@ -67,7 +67,7 @@ func run() error { providerBackend := gateway.NewNativeApolloBackend(providerManagement, providerRTSPAddress, providerRTSPURL, &http.Client{Timeout: 5 * time.Second}) provider := gateway.NewApolloAdapter(providerBackend, expectedIdentity) capabilities := gateway.DefaultCapabilities() - server, err := gateway.NewServer(gateway.ServerConfig{ListenAddress: listen, TLSConfig: serverTLS, GatewayID: gatewayID, Capabilities: capabilities, ProviderCapabilities: capabilities, Admission: controlPlaneClient, Provider: provider, PacerKbps: 100000}) + server, err := gateway.NewServer(gateway.ServerConfig{ListenAddress: listen, TLSConfig: serverTLS, GatewayID: gatewayID, Capabilities: capabilities, ProviderCapabilities: capabilities, Admission: controlPlaneClient, ProviderStateReporter: controlPlaneClient, Provider: provider, PacerKbps: 100000}) if err != nil { return err } diff --git a/gateway/control_plane.go b/gateway/control_plane.go index e7e01aa..25c8b19 100644 --- a/gateway/control_plane.go +++ b/gateway/control_plane.go @@ -76,6 +76,15 @@ func (c *ControlPlaneClient) Release(ctx context.Context, authority protocol.Ses return err } +func (c *ControlPlaneClient) ReportProviderState(ctx context.Context, state protocol.ProviderState) error { + payload, err := protocol.EncodeProviderState(state) + if err != nil { + return err + } + _, err = c.post(ctx, "/api/v1/gateway/provider-state", payload) + return err +} + func (c *ControlPlaneClient) post(ctx context.Context, path string, payload []byte) ([]byte, error) { if c == nil || c.HTTPClient == nil || c.BaseURL == "" { return nil, errors.New("control-plane client is not configured") diff --git a/gateway/gateway_test.go b/gateway/gateway_test.go index f408f1a..e5733e2 100644 --- a/gateway/gateway_test.go +++ b/gateway/gateway_test.go @@ -14,6 +14,7 @@ import ( "net" "os" "strings" + "sync" "sync/atomic" "testing" "time" @@ -238,7 +239,8 @@ func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) { fake := NewFakeApollo(FakeApolloConfig{Now: time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC)}) authority := protocol.SessionAuthority{Version: "1", SessionID: "session-1", GatewayID: "gateway-1", Audience: "versevdi-gateway", ReconnectSequence: 0, ExpiresAt: time.Now().Add(5 * time.Second).UTC().Format(time.RFC3339Nano), Capabilities: DefaultCapabilities(), ProviderProfile: ProviderProfileApollo, ProviderIdentity: fake.config.Identity.Key()} admission := &oneTimeAdmission{authority: authority, released: make(chan struct{})} - server, err := NewServer(ServerConfig{ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1", Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), Admission: admission, Provider: fake}) + reporter := &recordingProviderStateReporter{} + server, err := NewServer(ServerConfig{ListenAddress: "127.0.0.1:0", TLSConfig: serverTLS, GatewayID: "gateway-1", Capabilities: DefaultCapabilities(), ProviderCapabilities: DefaultCapabilities(), Admission: admission, ProviderStateReporter: reporter, Provider: fake}) if err != nil { t.Fatal(err) } @@ -246,7 +248,7 @@ func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) { defer cancel() serveDone := make(chan error, 1) go func() { serveDone <- server.Serve(ctx) }() - request := protocol.TunnelAdmissionRequest{Version: "1", SessionID: "session-1", GatewayID: "gateway-1", Audience: "versevdi-gateway", Grant: strings.Repeat("g", 64), ReconnectSequence: 0, ClientNonce: "nonce-0000000001", Capabilities: DefaultCapabilities()} + request := protocol.TunnelAdmissionRequest{Version: "1", SessionID: "session-1", GatewayID: "gateway-1", Audience: "versevdi-gateway", Grant: strings.Repeat("g", 64), ReconnectSequence: 0, ClientNonce: "nonce-0000000001", DeviceSignature: strings.Repeat("s", 86), Capabilities: DefaultCapabilities()} client, err := Dial(context.Background(), server.Addr().String(), clientTLS, request) if err != nil { _ = server.Close() @@ -282,6 +284,10 @@ func TestAdmissionQUICMTLSRelayAndCleanup(t *testing.T) { if err := <-serveDone; err != nil { t.Fatal(err) } + states := reporter.States() + if len(states) != 3 || states[0].State != ProviderStateStarting || states[1].State != ProviderStateReady || states[2].State != ProviderStateTerminated { + t.Fatalf("provider states = %#v", states) + } } func testTLS(t *testing.T) (*tls.Config, *tls.Config) { @@ -322,6 +328,24 @@ type oneTimeAdmission struct { released chan struct{} } +type recordingProviderStateReporter struct { + mu sync.Mutex + states []protocol.ProviderState +} + +func (r *recordingProviderStateReporter) ReportProviderState(_ context.Context, state protocol.ProviderState) error { + r.mu.Lock() + defer r.mu.Unlock() + r.states = append(r.states, state) + return nil +} + +func (r *recordingProviderStateReporter) States() []protocol.ProviderState { + r.mu.Lock() + defer r.mu.Unlock() + return append([]protocol.ProviderState(nil), r.states...) +} + func (a *oneTimeAdmission) Admit(context.Context, protocol.TunnelAdmissionRequest) (protocol.SessionAuthority, error) { if !a.used.CompareAndSwap(false, true) { return protocol.SessionAuthority{}, ErrAdmissionRejected diff --git a/gateway/provider.go b/gateway/provider.go index 073615c..c7fe839 100644 --- a/gateway/provider.go +++ b/gateway/provider.go @@ -338,12 +338,12 @@ func (f *FakeApollo) Setup(context.Context, LaunchRequest) ([]byte, error) { return []byte("RTSP/1.0 200 OK\r\nSession: fixture-session\r\nTransport: RTP/AVP/TCP;interleaved=0-1\r\n\r\n"), nil } -func (f *FakeApollo) Open(context.Context, LaunchRequest, RTSPResponse) (ProviderSession, error) { +func (f *FakeApollo) Open(_ context.Context, request LaunchRequest, _ RTSPResponse) (ProviderSession, error) { session := &fakeSession{ failure: f.config.Failure, video: make(chan []byte, 16), audio: make(chan []byte, 16), - state: protocol.ProviderState{Version: "1", State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, + state: protocol.ProviderState{Version: "1", SessionID: request.SessionID, State: ProviderStateStarting, Channels: []string{"video", "audio", "input", "feedback"}}, pressed: make(map[string]struct{}), } for _, payload := range f.config.Video { diff --git a/gateway/transport.go b/gateway/transport.go index 90d37b6..39c9a6d 100644 --- a/gateway/transport.go +++ b/gateway/transport.go @@ -44,18 +44,23 @@ func (f AdmissionFunc) Admit(ctx context.Context, request protocol.TunnelAdmissi func (AdmissionFunc) Release(context.Context, protocol.SessionAuthority) error { return nil } +type ProviderStateReporter interface { + ReportProviderState(context.Context, protocol.ProviderState) error +} + type ServerConfig struct { - ListenAddress string - TLSConfig *tls.Config - QUICConfig *quic.Config - GatewayID string - Capabilities protocol.CapabilityProfile - ProviderCapabilities protocol.CapabilityProfile - Admission Admission - Provider Provider - ProviderProfile string - ProviderIdentity string - PacerKbps int64 + ListenAddress string + TLSConfig *tls.Config + QUICConfig *quic.Config + GatewayID string + Capabilities protocol.CapabilityProfile + ProviderCapabilities protocol.CapabilityProfile + Admission Admission + ProviderStateReporter ProviderStateReporter + Provider Provider + ProviderProfile string + ProviderIdentity string + PacerKbps int64 } type Server struct { @@ -212,13 +217,26 @@ func (s *Server) handleConnection(parent context.Context, connection *quic.Conn) _ = writeStableError(stream, "no_capability_overlap", err, false) return } + if err := s.reportProviderState(ctx, protocol.ProviderState{Version: "1", SessionID: request.SessionID, State: ProviderStateStarting, CleanupPending: false, Channels: []string{"video", "audio", "input", "feedback"}}); err != nil { + _ = s.config.Admission.Release(context.Background(), authority) + _ = writeStableError(stream, "provider_state_unavailable", err, true) + return + } providerSession, err := s.config.Provider.Start(ctx, LaunchRequest{SessionID: request.SessionID, Capabilities: selected, ProviderProfile: authority.ProviderProfile, ProviderIdentity: authority.ProviderIdentity}) if err != nil { s.metrics.ProviderErrors.Add(1) + _ = s.reportProviderState(context.Background(), protocol.ProviderState{Version: "1", SessionID: request.SessionID, State: ProviderStateFailed, CleanupPending: false, Channels: []string{"video", "audio", "input", "feedback"}}) _ = s.config.Admission.Release(context.Background(), authority) _ = writeStableError(stream, stableProviderCode(err), err, errors.Is(err, context.DeadlineExceeded)) return } + if err := s.reportProviderState(ctx, providerSession.State()); err != nil { + _ = providerSession.ReleaseAll(context.Background()) + _ = providerSession.Terminate(context.Background()) + _ = s.config.Admission.Release(context.Background(), authority) + _ = writeStableError(stream, "provider_state_unavailable", err, true) + return + } authority.Capabilities = selected authorityBytes, err := protocol.EncodeSessionAuthority(authority) if err != nil || writeWire(stream, authorityBytes, defaultHelloLimit) != nil { @@ -237,6 +255,16 @@ func (s *Server) handleConnection(parent context.Context, connection *quic.Conn) session.run() } +func (s *Server) reportProviderState(ctx context.Context, state protocol.ProviderState) error { + if s.config.ProviderStateReporter == nil { + return nil + } + if err := state.Validate(); err != nil { + return err + } + return s.config.ProviderStateReporter.ReportProviderState(ctx, state) +} + func (s *Server) validateAuthority(authority protocol.SessionAuthority, request protocol.TunnelAdmissionRequest) error { if err := authority.Validate(); err != nil { return err @@ -467,15 +495,32 @@ func (s *gatewaySession) cleanup() { s.cancel() cleanupCtx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() - if err := s.provider.ReleaseAll(cleanupCtx); err != nil { + releaseInputsErr := s.provider.ReleaseAll(cleanupCtx) + if releaseInputsErr != nil { s.server.metrics.ProviderErrors.Add(1) } - if err := s.provider.Terminate(cleanupCtx); err != nil { + terminateErr := s.provider.Terminate(cleanupCtx) + if terminateErr != nil { s.server.metrics.ProviderErrors.Add(1) } + state := s.provider.State() + if releaseInputsErr != nil || terminateErr != nil { + state.State = ProviderStateCleanup + state.CleanupPending = true + if err := s.server.reportProviderState(cleanupCtx, state); err != nil { + s.server.metrics.ProviderErrors.Add(1) + } else if err := s.server.config.Admission.Release(cleanupCtx, s.authority); err != nil { + s.server.metrics.ProviderErrors.Add(1) + } + _ = s.connection.CloseWithError(applicationError, "session closed") + return + } if err := s.server.config.Admission.Release(cleanupCtx, s.authority); err != nil { s.server.metrics.ProviderErrors.Add(1) } + if err := s.server.reportProviderState(cleanupCtx, state); err != nil { + s.server.metrics.ProviderErrors.Add(1) + } _ = s.connection.CloseWithError(applicationError, "session closed") }) } diff --git a/go.mod b/go.mod index e7a6a3e..6c0f698 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module git.sechmachine.io.vn/sechmachine/VerseVDI-Data-Plane go 1.26.5 require ( - git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.1 + git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.3 github.com/quic-go/quic-go v0.61.0 ) diff --git a/go.sum b/go.sum index 15ff87e..68d8533 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.1 h1:RPpbmiXBED6Ry1mU/+OQuPdCCoXfn1jxvbZQBrlIRzQ= -git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.1/go.mod h1:7PhFIDhjtr20btWoEb2GqB+7dBpzJt43olrnHVutWoc= +git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.3 h1:ZoXbg9CRwlypVbDO0EaXwHVOKTGlIfZDC7s/4JuOISE= +git.sechmachine.io.vn/sechmachine/VerseVDI-Protocol v1.0.0-phase3c-gateway-rc.3/go.mod h1:7PhFIDhjtr20btWoEb2GqB+7dBpzJt43olrnHVutWoc= 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/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=