Add --log-out-of-order-messages flag

Author: mkuratczyk

README.md

@@ -88,6 +88,7 @@ messages published with perf-test can be consumed by `omq` or vice versa, and th
       --consumer-uri string                 URI for consuming
   -y, --consumers int                       The number of consumers to start (default 1)
   -h, --help                                help for omq
+      --log-out-of-order-messages           Print a log line when a message is received that is older than the previously received message
   -d, --message-durability                  Mark messages as durable (default true)
       --message-priority string             Message priority (0-255, default=unset)
       --metric-tags strings                 Prometheus label-value pairs, eg. l1=v1,l2=v2

cmd/root.go

@@ -213,6 +213,8 @@ func RootCmd() *cobra.Command {
 	rootCmd.PersistentFlags().IntVar(&cfg.ConsumerCredits, "consumer-credits", 1, "AMQP-1.0 consumer credits / STOMP prefetch count")
 	rootCmd.PersistentFlags().DurationVarP(&cfg.ConsumerLatency, "consumer-latency", "L", 0*time.Second, "consumer latency (time to accept message)")
 	rootCmd.PersistentFlags().StringSliceVar(&metricTags, "metric-tags", []string{}, "Prometheus label-value pairs, eg. l1=v1,l2=v2")
+	rootCmd.PersistentFlags().
+		BoolVar(&cfg.LogOutOfOrder, "log-out-of-order-messages", false, "Print a log line when a message is received that is older than the previously received message")
 
 	rootCmd.AddCommand(amqp_amqp)
 	rootCmd.AddCommand(amqp_stomp)

pkg/amqp10_client/consumer.go

@@ -99,7 +99,7 @@ func (c Amqp10Consumer) Start(ctx context.Context, subscribed chan bool) {
 			timeSent, latency := utils.CalculateEndToEndLatency(&payload)
 			m.With(prometheus.Labels{"protocol": "amqp-1.0"}).Observe(latency.Seconds())
 
-			if timeSent.Before(previousMessageTimeSent) {
+			if c.Config.LogOutOfOrder && timeSent.Before(previousMessageTimeSent) {
 				metrics.MessagesConsumedOutOfOrder.With(prometheus.Labels{"protocol": "amqp-1.0", "priority": priority}).Inc()
 				log.Info("Out of order message received. This message was sent before the previous message", "this messsage", timeSent, "previous message", previousMessageTimeSent)
 			}

pkg/config/config.go

@@ -55,6 +55,7 @@ type Config struct {
 	MqttPublisher        MqttOptions
 	MqttConsumer         MqttOptions
 	MetricTags           map[string]string
+	LogOutOfOrder        bool
 }
 
 func NewConfig() Config {

pkg/mqtt_client/consumer.go

@@ -70,7 +70,7 @@ func (c MqttConsumer) Start(ctx context.Context, subscribed chan bool) {
 		timeSent, latency := utils.CalculateEndToEndLatency(&payload)
 		m.Observe(latency.Seconds())
 
-		if timeSent.Before(previousMessageTimeSent) {
+		if c.Config.LogOutOfOrder && timeSent.Before(previousMessageTimeSent) {
 			metrics.MessagesConsumedOutOfOrder.With(prometheus.Labels{"protocol": "mqtt"}).Inc()
 			log.Info("Out of order message received. This message was sent before the previous message", "this messsage", timeSent, "previous message", previousMessageTimeSent)
 		}

pkg/stomp_client/consumer.go

@@ -80,7 +80,7 @@ func (c StompConsumer) Start(ctx context.Context, subscribed chan bool) {
 
 			priority := msg.Header.Get("priority")
 
-			if timeSent.Before(previousMessageTimeSent) {
+			if c.Config.LogOutOfOrder && timeSent.Before(previousMessageTimeSent) {
 				metrics.MessagesConsumedOutOfOrder.With(prometheus.Labels{"protocol": "amqp-1.0", "priority": priority}).Inc()
 				log.Info("Out of order message received. This message was sent before the previous message", "this messsage", timeSent, "previous message", previousMessageTimeSent)
 			}