CPU high usages

[zxyao145]

We are using apisix seed, and after deploying the application for a period of time, the application will experience a CPU usage rate of 100%. It was found that this is because the watcher.handleWatch method will receive endless messages from empty etcd. Here is the log:

"level":"debug","time":"2023-04-17T15:31:15.808098376+08:00","caller":"storer/etcd.go:209","message":"etcd watch prefix[/SERVICES/apisix/routes] event: {{16855301504475757525 1118403134429716127 0 134530 {} [] 0} [] 459241403 true false <nil> }"}
{"level":"info","time":"2023-04-17T15:31:15.80901158+08:00","caller":"components/watcher.go:109","message":"Watcher handleWatch receive msgs from etcd: routes, 0, []"}
{"level":"info","time":"2023-04-17T15:31:15.80903039+08:00","caller":"components/watcher.go:109","message":"Watcher handleWatch receive msgs from etcd: routes, 0, []"}
// repeat output Watcher handleWatch receive msgs from etcd: routes, 0

Then we added the judgment that the message is empty in etcd.Watch:

// Watch for changes on a key
func (s *EtcdV3) Watch(ctx context.Context, prefix string) <-chan []*message.Message {
	eventChan := s.client.Watch(ctx, prefix, clientv3.WithPrefix())
	ch := make(chan []*message.Message, 1)

	go func() {
		defer close(ch)

		for event := range eventChan {
+			logger.Debugf("etcd watch prefix[%s] event: %v", prefix, event)
			msgs := make([]*message.Message, 0, 16)

			for _, ev := range event.Events {
+				logger.Infof("watch changed, key: %s, version: %d", key, ev.Kv.Version)
				// ...
			}

+			if len(msgs) > 0 {
+				logger.Infof("Watcher handleWatch receive msgs from etcd: %+v, %+v", len(msgs), msgs)
+				ch <- msgs
+			}
		}
	}()

	return ch
}

From the log, if len(msgs) > 0 is working, because it doesn't output "Watcher handleWatch receive msgs xxx" with caller storer/etcd.go

And my watcher.handleWatch looks like this:

func (w *Watcher) handleWatch(s *storer.GenericStore) {
	logger.Debugf("Watcher handleWatch: %+v", s)
	ch := s.Watch()

	for {
		select {
		case <-w.ctx.Done():
			logger.Infof("Watcher handleWatch exit")
			return
		case msgs := <-ch: // receive msgs from etcd
			wg := sync.WaitGroup{}
			wg.Add(len(msgs))
+			logger.Infof("Watcher handleWatch receive msgs from etcd: %+v, %+v", len(msgs), msgs)
			for _, msg := range msgs {
				w.sem <- struct{}{}
				tc := tracer.NewTracer()
				ctx, span := tc.NewSpanConsumerKind(tc.RootCtx)
				go w.handleValue(ctx, msg, &wg, s)
				span.End()
			}
			wg.Wait()
		}
	}
}

I don't know what happened, and looking forward to your reply, thanks.