remove logs from local debugging session

Signed-off-by: Anton Troshin <[email protected]>

Author: antontroshin

pkg/runtime/compstore/subscriptions.go

@@ -17,12 +17,9 @@ import (
 	subapi "github.com/dapr/dapr/pkg/apis/subscriptions/v2alpha1"
 	runtimev1pb "github.com/dapr/dapr/pkg/proto/runtime/v1"
 	rtpubsub "github.com/dapr/dapr/pkg/runtime/pubsub"
-	"github.com/dapr/kit/logger"
 	"github.com/dapr/kit/ptr"
 )
 
-var log = logger.NewLogger("dapr.runtime.compstore.subscriptions")
-
 type DeclarativeSubscription struct {
 	Comp *subapi.Subscription
 	*NamedSubscription
@@ -78,11 +75,9 @@ func (c *ComponentStore) AddDeclarativeSubscription(comp *subapi.Subscription, s
 }
 
 func (c *ComponentStore) AddStreamSubscription(comp *subapi.Subscription, connectionID rtpubsub.ConnectionID) error {
-	log.Warn("Lock AddStreamSubscription")
 	c.lock.Lock()
 	defer func() {
 		c.lock.Unlock()
-		log.Warn("Unlock AddStreamSubscription defer")
 	}()
 
 	sub := &DeclarativeSubscription{
@@ -105,11 +100,9 @@ func (c *ComponentStore) AddStreamSubscription(comp *subapi.Subscription, connec
 }
 
 func (c *ComponentStore) DeleteStreamSubscription(comp *subapi.Subscription) {
-	log.Warn("Lock DeleteStreamSubscription")
 	c.lock.Lock()
 	defer func() {
 		c.lock.Unlock()
-		log.Warn("Unlock DeleteStreamSubscription defer")
 	}()
 	streamingSubscriptions, ok := c.subscriptions.streams[comp.Name]
 	if ok && len(streamingSubscriptions) > 0 {

pkg/runtime/processor/subscriber/subscriber.go

@@ -22,7 +22,6 @@ import (
 
 	"google.golang.org/grpc"
 
-	apierrors "github.com/dapr/dapr/pkg/api/errors"
 	"github.com/dapr/dapr/pkg/api/grpc/manager"
 	subapi "github.com/dapr/dapr/pkg/apis/subscriptions/v2alpha1"
 	"github.com/dapr/dapr/pkg/config"
@@ -128,25 +127,23 @@ func (s *Subscriber) ReloadPubSub(name string) error {
 }
 
 func (s *Subscriber) StartStreamerSubscription(subscription *subapi.Subscription, connectionID rtpubsub.ConnectionID) error {
-	log.Warn("Lock StartStreamerSubscription")
 	s.lock.Lock()
 	defer func() {
 		s.lock.Unlock()
-		log.Warn("Unlock StartStreamerSubscription defer")
 	}()
 
 	if s.closed {
-		return apierrors.PubSub("").WithMetadata(nil).DeserializeError(errors.New("subscriber is closed"))
+		return fmt.Errorf("streaming subscriber %s with ID %d is closed", subscription.Name, connectionID)
 	}
 
 	sub, found := s.compStore.GetStreamSubscription(subscription)
 	if !found {
-		return apierrors.PubSub("").WithMetadata(nil).NotFound()
+		return fmt.Errorf("streaming subscription %s not found", subscription.Name)
 	}
 
 	pubsub, ok := s.compStore.GetPubSub(sub.PubsubName)
 	if !ok {
-		return apierrors.PubSub(sub.PubsubName).WithMetadata(nil).NotFound()
+		return fmt.Errorf("streaming subscription pubsub %s not found", sub.PubsubName)
 	}
 
 	ss, err := s.startSubscription(pubsub, sub, true)
@@ -169,11 +166,9 @@ func (s *Subscriber) StartStreamerSubscription(subscription *subapi.Subscription
 }
 
 func (s *Subscriber) StopStreamerSubscription(subscription *subapi.Subscription, connectionID rtpubsub.ConnectionID) {
-	log.Warn("Lock StopStreamerSubscription")
 	s.lock.Lock()
 	defer func() {
 		s.lock.Unlock()
-		log.Warn("Unlock StopStreamerSubscription defer")
 	}()
 
 	if s.closed {
@@ -315,11 +310,9 @@ func (s *Subscriber) StopAppSubscriptions() {
 }
 
 func (s *Subscriber) StopAllSubscriptionsForever() {
-	log.Warn("Lock StopAllSubscriptionsForever")
 	s.lock.Lock()
 	defer func() {
 		s.lock.Unlock()
-		log.Warn("Unlock StopAllSubscriptionsForever defer")
 	}()
 
 	s.closed = true

pkg/runtime/pubsub/streamer/conn.go

@@ -33,7 +33,6 @@ type conn struct {
 
 func (c *conn) registerPublishResponse(id string) (chan *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1, func()) {
 	ch := make(chan *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1, 1)
-	log.Warnf("Lock registerPublishResponse messageId %s ConnectionID%d", id, c.connectionID)
 	c.lock.Lock()
 
 	if c.publishResponses3[id] == nil {
@@ -42,9 +41,7 @@ func (c *conn) registerPublishResponse(id string) (chan *rtv1pb.SubscribeTopicEv
 	c.publishResponses3[id][c.connectionID] = ch
 
 	c.lock.Unlock()
-	log.Warnf("Unlock registerPublishResponse messageId %s ConnectionID%d", id, c.connectionID)
 	return ch, func() {
-		log.Warnf("Lock registerPublishResponse defer messageId %s ConnectionID%d", id, c.connectionID)
 		c.lock.Lock()
 
 		delete(c.publishResponses3[id], c.connectionID)
@@ -59,16 +56,13 @@ func (c *conn) registerPublishResponse(id string) (chan *rtv1pb.SubscribeTopicEv
 		default:
 		}
 		close(ch)
-		log.Warnf("Unlock registerPublishResponse defer messageId %s ConnectionID%d", id, c.connectionID)
 	}
 }
 
 func (c *conn) notifyPublishResponse(resp *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1) {
-	log.Warnf("Lock notifyPublishResponse messageId %s ConnectionID%d", resp.GetId(), c.connectionID)
 	c.lock.RLock()
 	ch, ok := c.publishResponses3[resp.GetId()][c.connectionID]
 	c.lock.RUnlock()
-	log.Warnf("Unlock notifyPublishResponse messageId %s ConnectionID%d", resp.GetId(), c.connectionID)
 
 	if !ok {
 		log.Errorf("no client stream expecting publish response for id %s ConnectionID%d", resp.GetId(), c.connectionID)

pkg/runtime/pubsub/streamer/streamer.go

@@ -59,7 +59,6 @@ func New(opts Options) rtpubsub.AdapterStreamer {
 }
 
 func (s *streamer) Subscribe(stream rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server, req *rtv1pb.SubscribeTopicEventsRequestInitialAlpha1, connectionID rtpubsub.ConnectionID) error {
-	log.Warn("Lock Subscribe ConnectionID", connectionID)
 	s.lock.Lock()
 	key := s.StreamerKey(req.GetPubsubName(), req.GetTopic())
 
@@ -74,12 +73,10 @@ func (s *streamer) Subscribe(stream rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server
 	}
 	s.subscribers[key][connectionID] = connection
 
-	log.Infof("Subscribing to pubsub '%s' topic '%s' ConnectionID%d", req.GetPubsubName(), req.GetTopic(), connectionID)
+	log.Infof("Subscribing to pubsub '%s' topic '%s' ConnectionID %d", req.GetPubsubName(), req.GetTopic(), connectionID)
 	s.lock.Unlock()
-	log.Warn("Unlock Subscribe ConnectionID", connectionID)
 
 	defer func() {
-		log.Warn("Lock Subscribe defer")
 		s.lock.Lock()
 		if connections := s.subscribers[key]; connections != nil {
 			delete(connections, connectionID)
@@ -88,7 +85,6 @@ func (s *streamer) Subscribe(stream rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server
 			}
 		}
 		s.lock.Unlock()
-		log.Warn("Unlock Subscribe defer")
 	}()
 
 	errCh := make(chan error, 2)
@@ -103,15 +99,12 @@ func (s *streamer) Subscribe(stream rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server
 
 	go func() {
 		for {
-			log.Info("Waiting for Recv, connectionID", connectionID)
 			resp, err := stream.Recv()
-			log.Infof("Recv, connectionID%d, resp%s", connectionID, resp)
 			s, ok := status.FromError(err)
 
 			if (ok && s.Code() == codes.Canceled) ||
 				errors.Is(err, context.Canceled) ||
 				errors.Is(err, io.EOF) {
-				log.Infof("Unsubscribed from pubsub '%s' topic '%s'", req.GetPubsubName(), req.GetTopic())
 				errCh <- err
 				return
 			}
@@ -143,12 +136,10 @@ func (s *streamer) Subscribe(stream rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server
 }
 
 func (s *streamer) Publish(ctx context.Context, msg *rtpubsub.SubscribedMessage) error {
-	log.Warn("RLock Publish ConnectionID", msg.SubscriberID)
 	s.lock.RLock()
 	key := s.StreamerKey(msg.PubSub, msg.Topic)
 	connection, ok := s.subscribers[key][msg.SubscriberID]
 	s.lock.RUnlock()
-	log.Warn("RUnlock Publish ConnectionID", msg.SubscriberID)
 	if !ok {
 		return fmt.Errorf("no streamer subscribed to pubsub %q topic %q", msg.PubSub, msg.Topic)
 	}
@@ -169,15 +160,13 @@ func (s *streamer) Publish(ctx context.Context, msg *rtpubsub.SubscribedMessage)
 	defer cleanup()
 
 	start := time.Now()
-	log.Warn("Lock stream ConnectionID", connection.connectionID)
 	connection.streamLock.Lock()
 	err = connection.stream.Send(&rtv1pb.SubscribeTopicEventsResponseAlpha1{
 		SubscribeTopicEventsResponseType: &rtv1pb.SubscribeTopicEventsResponseAlpha1_EventMessage{
 			EventMessage: envelope,
 		},
 	})
 	connection.streamLock.Unlock()
-	log.Warn("Unlock stream ConnectionID", connection.connectionID)
 	elapsed := diag.ElapsedSince(start)
 
 	if span != nil {
@@ -227,11 +216,9 @@ func (s *streamer) StreamerKey(pubsub, topic string) string {
 }
 
 func (s *streamer) Close(key string, connectionID rtpubsub.ConnectionID) {
-	log.Warn("RLock Close")
 	s.lock.RLock()
 	defer func() {
 		s.lock.RUnlock()
-		log.Warn("Unlock Close defer")
 	}()
 
 	if conn, ok := s.subscribers[key][connectionID]; ok {

pkg/runtime/subscription/subscription.go

@@ -122,7 +122,6 @@ func New(opts Options) (*Subscription, error) {
 		Topic:    subscribeTopic,
 		Metadata: routeMetadata,
 	}, func(ctx context.Context, msg *contribpubsub.NewMessage) error {
-		log.Warnf("Handler Message ConnectionID%d message %s", s.connectionID, msg.Data)
 		if msg.Metadata == nil {
 			msg.Metadata = make(map[string]string, 1)
 		}