Adopt updated ingress RPC API (#1068)

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

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=

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)
 			}

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
 }
 

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")
 	}

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) {

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