From 8cf0872ec1f4345483a9891db9f5fccd4329db3f Mon Sep 17 00:00:00 2001 From: Benjamin Pracht Date: Mon, 3 Oct 2022 11:31:55 -0700 Subject: [PATCH] Adopt updated ingress RPC API (#1068) --- go.mod | 2 +- go.sum | 4 +- pkg/service/ingress.go | 32 +-- pkg/service/interfaces.go | 1 + pkg/service/redisstore.go | 186 +++++++++++++++--- pkg/service/redisstore_test.go | 11 +- .../servicefakes/fake_ingress_store.go | 78 ++++++++ 7 files changed, 262 insertions(+), 52 deletions(-) diff --git a/go.mod b/go.mod index 02a99c255..b97c1b9db 100644 --- a/go.mod +++ b/go.mod @@ -17,7 +17,7 @@ require ( github.com/hashicorp/go-version v1.6.0 github.com/hashicorp/golang-lru v0.5.4 github.com/livekit/mageutil v0.0.0-20221002073820-d9198083cfdc - github.com/livekit/protocol v1.1.3-0.20220926180122-bafacf7041a5 + github.com/livekit/protocol v1.1.3-0.20221003181058-44a5ed6b4d13 github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 github.com/mackerelio/go-osstat v0.2.3 github.com/magefile/mage v1.14.0 diff --git a/go.sum b/go.sum index eef73c893..7caceb385 100644 --- a/go.sum +++ b/go.sum @@ -242,8 +242,8 @@ github.com/lithammer/shortuuid/v3 v3.0.7 h1:trX0KTHy4Pbwo/6ia8fscyHoGA+mf1jWbPJV github.com/lithammer/shortuuid/v3 v3.0.7/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= github.com/livekit/mageutil v0.0.0-20221002073820-d9198083cfdc h1:e3GIA9AL6h4a38MLHCwTTKzW/JCIGTtpwavPj82Tcfo= github.com/livekit/mageutil v0.0.0-20221002073820-d9198083cfdc/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= -github.com/livekit/protocol v1.1.3-0.20220926180122-bafacf7041a5 h1:ct0ellNBvJ67i1VSxI9NIYAypBGRBdaJqucVm5PWIE0= -github.com/livekit/protocol v1.1.3-0.20220926180122-bafacf7041a5/go.mod h1:eburCdz6ZtbgKSKYkAeCdWP1z33DB9clTphz7uNaxp0= +github.com/livekit/protocol v1.1.3-0.20221003181058-44a5ed6b4d13 h1:bmJcE0DiwCVESGS5LWMkWb+yIg8uCAtlbf5+o7l6+vM= +github.com/livekit/protocol v1.1.3-0.20221003181058-44a5ed6b4d13/go.mod h1:eburCdz6ZtbgKSKYkAeCdWP1z33DB9clTphz7uNaxp0= github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 h1:vOaY2qvfLihDyeZtnGGN1Law9wRrw8BMGCr1TygTvMw= github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995/go.mod h1:116ych8UaEs9vfIE8n6iZCZ30iagUFTls0vRmC+Ix5U= github.com/mackerelio/go-osstat v0.2.3 h1:jAMXD5erlDE39kdX2CU7YwCGRcxIO33u/p8+Fhe5dJw= diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index 854cf4712..41bb25219 100644 --- a/pkg/service/ingress.go +++ b/pkg/service/ingress.go @@ -86,9 +86,6 @@ func (s *IngressService) CreateIngressWithUrlPrefix(ctx context.Context, urlPref ParticipantIdentity: req.ParticipantIdentity, ParticipantName: req.ParticipantName, Reusable: req.InputType == livekit.IngressInput_RTMP_INPUT, - State: &livekit.IngressState{ - Status: livekit.IngressState_ENDPOINT_INACTIVE, - }, } if err := s.store.StoreIngress(ctx, info); err != nil { @@ -99,10 +96,10 @@ func (s *IngressService) CreateIngressWithUrlPrefix(ctx context.Context, urlPref return info, nil } -func (s *IngressService) sendRPCWithRetry(ctx context.Context, req *livekit.IngressRequest) (*livekit.IngressInfo, error) { +func (s *IngressService) sendRPCWithRetry(ctx context.Context, req *livekit.IngressRequest) (*livekit.IngressState, error) { type result struct { - info *livekit.IngressInfo - err error + state *livekit.IngressState + err error } resChan := make(chan result, 1) @@ -118,9 +115,9 @@ func (s *IngressService) sendRPCWithRetry(ctx context.Context, req *livekit.Ingr default: } - i, err := s.rpcClient.SendRequest(cctx, req) + s, err := s.rpcClient.SendRequest(cctx, req) if err != ingress.ErrNoResponse { - resChan <- result{i, err} + resChan <- result{s, err} return } } @@ -128,7 +125,7 @@ func (s *IngressService) sendRPCWithRetry(ctx context.Context, req *livekit.Ingr select { case res := <-resChan: - return res.info, res.err + return res.state, res.err case <-time.After(initialTimeout): return nil, ingress.ErrNoResponse } @@ -156,6 +153,10 @@ func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateI switch info.State.Status { case livekit.IngressState_ENDPOINT_ERROR: info.State.Status = livekit.IngressState_ENDPOINT_INACTIVE + err = s.store.UpdateIngressState(ctx, req.IngressId, info.State) + if err != nil { + logger.Warnw("could not store ingress state", err) + } fallthrough case livekit.IngressState_ENDPOINT_INACTIVE: @@ -180,14 +181,15 @@ func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateI case livekit.IngressState_ENDPOINT_BUFFERING, livekit.IngressState_ENDPOINT_PUBLISHING: - i, err := s.sendRPCWithRetry(ctx, &livekit.IngressRequest{ + // Do not update store the returned state as the ingress service will do it + s, err := s.sendRPCWithRetry(ctx, &livekit.IngressRequest{ IngressId: req.IngressId, Request: &livekit.IngressRequest_Update{Update: req}, }) if err != nil { logger.Warnw("could not update active ingress", err) } else { - info = i + info.State = s } } @@ -235,14 +237,14 @@ func (s *IngressService) DeleteIngress(ctx context.Context, req *livekit.DeleteI switch info.State.Status { case livekit.IngressState_ENDPOINT_BUFFERING, livekit.IngressState_ENDPOINT_PUBLISHING: - i, err := s.sendRPCWithRetry(ctx, &livekit.IngressRequest{ + s, err := s.sendRPCWithRetry(ctx, &livekit.IngressRequest{ IngressId: req.IngressId, Request: &livekit.IngressRequest_Delete{Delete: req}, }) if err != nil { logger.Warnw("could not stop active ingress", err) } else { - info = i + info.State = s } } @@ -269,14 +271,14 @@ func (s *IngressService) updateWorker() { case msg := <-resChan: b := sub.Payload(msg) - res := &livekit.IngressInfo{} + res := &livekit.UpdateIngressStateRequest{} if err = proto.Unmarshal(b, res); err != nil { logger.Errorw("failed to read results", err) continue } // save updated info to store - err = s.store.UpdateIngress(context.Background(), res) + err = s.store.UpdateIngressState(context.Background(), res.IngressId, res.State) if err != nil { logger.Errorw("could not update ingress", err) } diff --git a/pkg/service/interfaces.go b/pkg/service/interfaces.go index 88e051770..c2b9876e6 100644 --- a/pkg/service/interfaces.go +++ b/pkg/service/interfaces.go @@ -53,6 +53,7 @@ type IngressStore interface { LoadIngressFromStreamKey(ctx context.Context, streamKey string) (*livekit.IngressInfo, error) ListIngress(ctx context.Context, roomName livekit.RoomName) ([]*livekit.IngressInfo, error) UpdateIngress(ctx context.Context, info *livekit.IngressInfo) error + UpdateIngressState(ctx context.Context, ingressId string, state *livekit.IngressState) error DeleteIngress(ctx context.Context, info *livekit.IngressInfo) error } diff --git a/pkg/service/redisstore.go b/pkg/service/redisstore.go index d022b3711..4d5c366f5 100644 --- a/pkg/service/redisstore.go +++ b/pkg/service/redisstore.go @@ -33,9 +33,10 @@ const ( DeprecatedRoomEgressPrefix = "room_egress:" // IngressKey is a hash of ingressID => ingress info - IngressKey = "ingress" - StreamKeyKey = "stream_key" - RoomIngressPrefix = "room_ingress:" + IngressKey = "ingress" + StreamKeyKey = "stream_key" + IngressStatePrefix = "ingress_state:" + RoomIngressPrefix = "room_ingress:" // RoomParticipantsPrefix is hash of participant_name => ParticipantInfo RoomParticipantsPrefix = "room_participants:" @@ -476,10 +477,15 @@ func parseEgressEnded(value string) (roomName string, endedAt int64, err error) } func (s *RedisStore) StoreIngress(ctx context.Context, info *livekit.IngressInfo) error { - return s.storeIngress(ctx, info, false) + err := s.storeIngress(ctx, info) + if err != nil { + return err + } + + return s.storeIngressState(ctx, info.IngressId, nil) } -func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo, updateOnly bool) error { +func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo) error { if info.IngressId == "" { return errors.New("Missing IngressId") } @@ -487,11 +493,12 @@ func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo, return errors.New("Missing StreamKey") } - if info.State == nil { - info.State = &livekit.IngressState{} - } + // ignore state + infoCopy := livekit.IngressInfo{} + infoCopy = *info + infoCopy.State = nil - data, err := proto.Marshal(info) + data, err := proto.Marshal(&infoCopy) if err != nil { return err } @@ -499,28 +506,18 @@ func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo, // Use a "transaction" to remove the old room association if it changed txf := func(tx *redis.Tx) error { var oldRoom string - var oldStartedAt int64 oldInfo, err := s.loadIngress(tx, info.IngressId) switch err { case ErrIngressNotFound: // Ingress doesn't exist yet - if updateOnly { - return err - } case nil: oldRoom = oldInfo.RoomName - oldStartedAt = oldInfo.State.StartedAt default: return err } results, err := tx.TxPipelined(s.ctx, func(p redis.Pipeliner) error { - if info.State.StartedAt < oldStartedAt { - // Do not overwrite the info and state of a more recent session - return ingress.ErrIngressOutOfDate - } - p.HSet(s.ctx, IngressKey, info.IngressId, data) p.HSet(s.ctx, StreamKeyKey, info.StreamKey, info.IngressId) @@ -551,7 +548,80 @@ func (s *RedisStore) storeIngress(_ context.Context, info *livekit.IngressInfo, // Retry if the key has been changed. for i := 0; i < maxRetries; i++ { - err := s.rc.Watch(s.ctx, txf, IngressKey, StreamKeyKey) + err := s.rc.Watch(s.ctx, txf, IngressKey) + switch err { + case redis.TxFailedErr: + // Optimistic lock lost. Retry. + continue + default: + return err + } + } + + return nil +} + +func (s *RedisStore) storeIngressState(_ context.Context, ingressId string, state *livekit.IngressState) error { + if ingressId == "" { + return errors.New("Missing IngressId") + } + + if state == nil { + state = &livekit.IngressState{} + } + + data, err := proto.Marshal(state) + if err != nil { + return err + } + + // Use a "transaction" to remove the old room association if it changed + txf := func(tx *redis.Tx) error { + var oldStartedAt int64 + + info, err := s.loadIngress(tx, ingressId) + if err != nil { + return err + } + + oldState, err := s.loadIngressState(tx, ingressId) + switch err { + case ErrIngressNotFound: + // Ingress state doesn't exist yet + case nil: + oldStartedAt = oldState.StartedAt + default: + return err + } + + results, err := tx.TxPipelined(s.ctx, func(p redis.Pipeliner) error { + if state.StartedAt < oldStartedAt { + // Do not overwrite the info and state of a more recent session + return ingress.ErrIngressOutOfDate + } + + p.Set(s.ctx, IngressStatePrefix+ingressId, data, 0) + p.HSet(s.ctx, StreamKeyKey, info.StreamKey, info.IngressId) + + return nil + }) + + if err != nil { + return err + } + + for _, res := range results { + if err := res.Err(); err != nil { + return err + } + } + + return nil + } + + // Retry if the key has been changed. + for i := 0; i < maxRetries; i++ { + err := s.rc.Watch(s.ctx, txf, IngressKey, IngressStatePrefix+ingressId) switch err { case redis.TxFailedErr: // Optimistic lock lost. Retry. @@ -583,15 +653,16 @@ func (s *RedisStore) loadIngress(c redis.Cmdable, ingressId string) (*livekit.In } } -func (s *RedisStore) LoadIngress(_ context.Context, ingressId string) (*livekit.IngressInfo, error) { - return s.loadIngress(s.rc, ingressId) -} - -func (s *RedisStore) LoadIngressFromStreamKey(_ context.Context, streamKey string) (*livekit.IngressInfo, error) { - ingressID, err := s.rc.HGet(s.ctx, StreamKeyKey, streamKey).Result() +func (s *RedisStore) loadIngressState(c redis.Cmdable, ingressId string) (*livekit.IngressState, error) { + data, err := c.Get(s.ctx, IngressStatePrefix+ingressId).Result() switch err { case nil: - return s.loadIngress(s.rc, ingressID) + state := &livekit.IngressState{} + err = proto.Unmarshal([]byte(data), state) + if err != nil { + return nil, err + } + return state, nil case redis.Nil: return nil, ErrIngressNotFound @@ -601,7 +672,39 @@ func (s *RedisStore) LoadIngressFromStreamKey(_ context.Context, streamKey strin } } -func (s *RedisStore) ListIngress(_ context.Context, roomName livekit.RoomName) ([]*livekit.IngressInfo, error) { +func (s *RedisStore) LoadIngress(_ context.Context, ingressId string) (*livekit.IngressInfo, error) { + info, err := s.loadIngress(s.rc, ingressId) + if err != nil { + return nil, err + } + state, err := s.loadIngressState(s.rc, ingressId) + switch err { + case nil: + info.State = state + case ErrIngressNotFound: + // No state for this ingress + default: + return nil, err + } + + return info, nil +} + +func (s *RedisStore) LoadIngressFromStreamKey(_ context.Context, streamKey string) (*livekit.IngressInfo, error) { + ingressID, err := s.rc.HGet(s.ctx, StreamKeyKey, streamKey).Result() + switch err { + case nil: + return s.LoadIngress(s.ctx, ingressID) + + case redis.Nil: + return nil, ErrIngressNotFound + + default: + return nil, err + } +} + +func (s *RedisStore) ListIngress(ctx context.Context, roomName livekit.RoomName) ([]*livekit.IngressInfo, error) { var infos []*livekit.IngressInfo if roomName == "" { @@ -619,6 +722,16 @@ func (s *RedisStore) ListIngress(_ context.Context, roomName livekit.RoomName) ( if err != nil { return nil, err } + state, err := s.loadIngressState(s.rc, info.IngressId) + switch err { + case nil: + info.State = state + case ErrIngressNotFound: + // No state for this ingress + default: + return nil, err + } + infos = append(infos, info) } } else { @@ -640,6 +753,16 @@ func (s *RedisStore) ListIngress(_ context.Context, roomName livekit.RoomName) ( if err != nil { return nil, err } + state, err := s.loadIngressState(s.rc, info.IngressId) + switch err { + case nil: + info.State = state + case ErrIngressNotFound: + // No state for this ingress + default: + return nil, err + } + infos = append(infos, info) } } @@ -648,7 +771,11 @@ func (s *RedisStore) ListIngress(_ context.Context, roomName livekit.RoomName) ( } func (s *RedisStore) UpdateIngress(ctx context.Context, info *livekit.IngressInfo) error { - return s.storeIngress(ctx, info, true) + return s.storeIngress(ctx, info) +} + +func (s *RedisStore) UpdateIngressState(ctx context.Context, ingressId string, state *livekit.IngressState) error { + return s.storeIngressState(ctx, ingressId, state) } func (s *RedisStore) DeleteIngress(_ context.Context, info *livekit.IngressInfo) error { @@ -656,6 +783,7 @@ func (s *RedisStore) DeleteIngress(_ context.Context, info *livekit.IngressInfo) tx.SRem(s.ctx, RoomIngressPrefix+info.RoomName, info.IngressId) tx.HDel(s.ctx, StreamKeyKey, info.IngressId) tx.HDel(s.ctx, IngressKey, info.IngressId) + tx.Del(s.ctx, IngressStatePrefix+info.IngressId) if _, err := tx.Exec(s.ctx); err != nil { return errors.Wrap(err, "could not delete ingress info") } diff --git a/pkg/service/redisstore_test.go b/pkg/service/redisstore_test.go index 07fae901b..15f60a679 100644 --- a/pkg/service/redisstore_test.go +++ b/pkg/service/redisstore_test.go @@ -251,6 +251,9 @@ func TestIngressStore(t *testing.T) { err := rs.StoreIngress(ctx, info) require.NoError(t, err) + err = rs.UpdateIngressState(ctx, info.IngressId, info.State) + require.NoError(t, err) + t.Cleanup(func() { rs.DeleteIngress(ctx, info) }) @@ -282,20 +285,18 @@ func TestIngressStore(t *testing.T) { require.NoError(t, err) require.Equal(t, 0, len(infos)) - info.RoomName = "room2" info.State.StartedAt = 1 - err = rs.UpdateIngress(ctx, info) + err = rs.UpdateIngressState(ctx, info.IngressId, info.State) require.Equal(t, ingress.ErrIngressOutOfDate, err) - info.RoomName = "room2" info.State.StartedAt = 3 - err = rs.UpdateIngress(ctx, info) + err = rs.UpdateIngressState(ctx, info.IngressId, info.State) require.NoError(t, err) infos, err = rs.ListIngress(ctx, "") require.NoError(t, err) require.Equal(t, 1, len(infos)) - require.Equal(t, "room2", infos[0].RoomName) + require.Equal(t, "", infos[0].RoomName) } func compareIngressInfo(t *testing.T, expected, v *livekit.IngressInfo) { diff --git a/pkg/service/servicefakes/fake_ingress_store.go b/pkg/service/servicefakes/fake_ingress_store.go index 542d92e90..4265e1d5b 100644 --- a/pkg/service/servicefakes/fake_ingress_store.go +++ b/pkg/service/servicefakes/fake_ingress_store.go @@ -88,6 +88,19 @@ type FakeIngressStore struct { updateIngressReturnsOnCall map[int]struct { result1 error } + UpdateIngressStateStub func(context.Context, string, *livekit.IngressState) error + updateIngressStateMutex sync.RWMutex + updateIngressStateArgsForCall []struct { + arg1 context.Context + arg2 string + arg3 *livekit.IngressState + } + updateIngressStateReturns struct { + result1 error + } + updateIngressStateReturnsOnCall map[int]struct { + result1 error + } invocations map[string][][]interface{} invocationsMutex sync.RWMutex } @@ -473,6 +486,69 @@ func (fake *FakeIngressStore) UpdateIngressReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeIngressStore) UpdateIngressState(arg1 context.Context, arg2 string, arg3 *livekit.IngressState) error { + fake.updateIngressStateMutex.Lock() + ret, specificReturn := fake.updateIngressStateReturnsOnCall[len(fake.updateIngressStateArgsForCall)] + fake.updateIngressStateArgsForCall = append(fake.updateIngressStateArgsForCall, struct { + arg1 context.Context + arg2 string + arg3 *livekit.IngressState + }{arg1, arg2, arg3}) + stub := fake.UpdateIngressStateStub + fakeReturns := fake.updateIngressStateReturns + fake.recordInvocation("UpdateIngressState", []interface{}{arg1, arg2, arg3}) + fake.updateIngressStateMutex.Unlock() + if stub != nil { + return stub(arg1, arg2, arg3) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeIngressStore) UpdateIngressStateCallCount() int { + fake.updateIngressStateMutex.RLock() + defer fake.updateIngressStateMutex.RUnlock() + return len(fake.updateIngressStateArgsForCall) +} + +func (fake *FakeIngressStore) UpdateIngressStateCalls(stub func(context.Context, string, *livekit.IngressState) error) { + fake.updateIngressStateMutex.Lock() + defer fake.updateIngressStateMutex.Unlock() + fake.UpdateIngressStateStub = stub +} + +func (fake *FakeIngressStore) UpdateIngressStateArgsForCall(i int) (context.Context, string, *livekit.IngressState) { + fake.updateIngressStateMutex.RLock() + defer fake.updateIngressStateMutex.RUnlock() + argsForCall := fake.updateIngressStateArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 +} + +func (fake *FakeIngressStore) UpdateIngressStateReturns(result1 error) { + fake.updateIngressStateMutex.Lock() + defer fake.updateIngressStateMutex.Unlock() + fake.UpdateIngressStateStub = nil + fake.updateIngressStateReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeIngressStore) UpdateIngressStateReturnsOnCall(i int, result1 error) { + fake.updateIngressStateMutex.Lock() + defer fake.updateIngressStateMutex.Unlock() + fake.UpdateIngressStateStub = nil + if fake.updateIngressStateReturnsOnCall == nil { + fake.updateIngressStateReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.updateIngressStateReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeIngressStore) Invocations() map[string][][]interface{} { fake.invocationsMutex.RLock() defer fake.invocationsMutex.RUnlock() @@ -488,6 +564,8 @@ func (fake *FakeIngressStore) Invocations() map[string][][]interface{} { defer fake.storeIngressMutex.RUnlock() fake.updateIngressMutex.RLock() defer fake.updateIngressMutex.RUnlock() + fake.updateIngressStateMutex.RLock() + defer fake.updateIngressStateMutex.RUnlock() copiedInvocations := map[string][][]interface{}{} for key, value := range fake.invocations { copiedInvocations[key] = value