Merge pull request #80 from planetary-social/rate-limit-from-notices

Rate limit from notice warnings

cmd/send-all-events-to-relay/main.go

@@ -339,3 +339,6 @@ func (m mockMetrics) ReportMessageReceived(address domain.RelayAddress, messageT
 
 func (m mockMetrics) ReportRelayDisconnection(address domain.RelayAddress, err error) {
 }
+
+func (m mockMetrics) ReportNotice(address domain.RelayAddress, noticeType relays.NoticeType) {
+}

internal/logging/log.go

@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"reflect"
 	"runtime"
+	"time"
 
 	"github.com/davecgh/go-spew/spew"
 )
@@ -202,6 +203,16 @@ func (d devNullLoggerEntry) WithField(key string, v any) Entry {
 func (d devNullLoggerEntry) Message(msg string) {
 }
 
+// logPeriodically executes the passed action function if the current time in milliseconds
+// modulo logInterval equals zero. This approach allows executing the action periodically,
+// approximating the execution to happen once every `logInterval` milliseconds.
+func LogPeriodically(action func(), logInterval int64) {
+	currentTimeMillis := time.Now().UnixNano() / int64(time.Millisecond)
+	if currentTimeMillis%logInterval == 0 {
+		action()
+	}
+}
+
 func Inspect(args ...interface{}) {
 	for _, arg := range args {
 		val := reflect.ValueOf(arg)

service/adapters/mocks/metrics.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/planetary-social/nos-event-service/service/app"
 	"github.com/planetary-social/nos-event-service/service/domain"
+	"github.com/planetary-social/nos-event-service/service/domain/relays"
 )
 
 type Metrics struct {
@@ -39,6 +40,9 @@ func (m Metrics) ReportNumberOfStoredEvents(n int) {
 func (m Metrics) ReportEventSentToRelay(address domain.RelayAddress, decision app.SendEventToRelayDecision, result app.SendEventToRelayResult) {
 }
 
+func (m Metrics) ReportNotice(address domain.RelayAddress, noticeType relays.NoticeType) {
+}
+
 type ApplicationCall struct {
 }
 

service/adapters/prometheus/prometheus.go

@@ -32,10 +32,12 @@ const (
 
 	labelReason = "reason"
 
-	labelDecision = "decision"
+	labelDecision   = "decision"
+	labelNoticeType = "noticeType"
 )
 
 type Prometheus struct {
+	noticeTypeCounter                       *prometheus.CounterVec
 	applicationHandlerCallsCounter          *prometheus.CounterVec
 	applicationHandlerCallDurationHistogram *prometheus.HistogramVec
 	relayDownloadersGauge                   prometheus.Gauge
@@ -56,6 +58,13 @@ type Prometheus struct {
 }
 
 func NewPrometheus(logger logging.Logger) (*Prometheus, error) {
+	noticeTypeCounter := prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "notice_type_total",
+			Help: "Total number of notices per notice type.",
+		},
+		[]string{labelAddress, labelNoticeType},
+	)
 	applicationHandlerCallsCounter := prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Name: "application_handler_calls_total",
@@ -154,6 +163,7 @@ func NewPrometheus(logger logging.Logger) (*Prometheus, error) {
 
 	reg := prometheus.NewRegistry()
 	for _, v := range []prometheus.Collector{
+		noticeTypeCounter,
 		applicationHandlerCallsCounter,
 		applicationHandlerCallDurationHistogram,
 		versionGague,
@@ -191,6 +201,7 @@ func NewPrometheus(logger logging.Logger) (*Prometheus, error) {
 	}
 
 	return &Prometheus{
+		noticeTypeCounter:                       noticeTypeCounter,
 		applicationHandlerCallsCounter:          applicationHandlerCallsCounter,
 		applicationHandlerCallDurationHistogram: applicationHandlerCallDurationHistogram,
 		relayDownloadersGauge:                   relayDownloadersGauge,
@@ -281,6 +292,13 @@ func (p *Prometheus) ReportEventSentToRelay(address domain.RelayAddress, decisio
 	}).Inc()
 }
 
+func (p *Prometheus) ReportNotice(address domain.RelayAddress, noticeType relays.NoticeType) {
+	p.noticeTypeCounter.With(prometheus.Labels{
+		labelAddress:    address.String(),
+		labelNoticeType: string(noticeType),
+	}).Inc()
+}
+
 func (p *Prometheus) Registry() *prometheus.Registry {
 	return p.registry
 }

service/domain/downloader/scheduler.go

@@ -372,7 +372,7 @@ func (t *TimeWindowTaskGenerator) Generate(ctx context.Context, kinds []domain.E
 func (t *TimeWindowTaskGenerator) maybeGenerateNewTracker(ctx context.Context) (*TimeWindowTaskTracker, bool, error) {
 	nextWindow := t.lastWindow.Advance()
 	now := t.currentTimeProvider.GetCurrentTime()
-	if nextWindow.End().After(now.Add(-time.Minute)) {
+	if nextWindow.End().After(now.Add(-windowSize)) {
 		return nil, false, nil
 	}
 	t.lastWindow = nextWindow