bugfix: avoid logging non-200 response and don't publish mesasge when…

… there is err

api/pkg/inference-logger/logger/mlobs_sink.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
+	"net/http"
 
 	"google.golang.org/protobuf/types/known/timestamppb"
 
@@ -168,16 +169,18 @@ func (m *MLObsSink) buildNewKafkaMessage(predictionLog *upiv1.PredictionLog) (*k
 
 func (m *MLObsSink) Sink(rawLogEntries []*LogEntry) error {
 	for _, rawLogEntry := range rawLogEntries {
-		if rand.Float64() >= SamplingRate {
+		if rawLogEntry.ResponsePayload.StatusCode != http.StatusOK || rand.Float64() >= SamplingRate {
 			continue
 		}
 		predictionLog, err := m.newPredictionLog(rawLogEntry)
 		if err != nil {
 			m.logger.Errorf("unable to convert log entry: %v", err)
+			continue
 		}
 		kafkaMessage, err := m.buildNewKafkaMessage(predictionLog)
 		if err != nil {
 			m.logger.Errorf("unable to build kafka message: %v", err)
+			continue
 		}
 		err = m.producer.Produce(kafkaMessage, m.producer.Events())
 		if err != nil {

python/observation-publisher/publisher/prediction_log_consumer.py

@@ -151,7 +151,7 @@ def poll_new_logs(self) -> List[PredictionLog]:
         return [
             parse_message_to_prediction_log(msg.value())
             for msg in messages
-            if (msg is not None and msg.error() is None)
+            if (msg is not None and msg.error() is None and msg.value() is not None)
         ]
 
     def commit(self):