fix panic on reading stream stats (#881)

* fix panic on reading stream stats

* avoid converting potentially incorrect values for now

Author: frostbyte73

pkg/pipeline/builder/stream.go

@@ -271,15 +271,31 @@ func (s *Stream) Stats() (*logging.StreamStats, bool) {
 	if err != nil || structure == nil {
 		return nil, false
 	}
+
 	if stats := structure.(*gst.Structure).Values(); stats != nil {
 		return &logging.StreamStats{
 			Timestamp:     time.Now().Format(time.DateTime),
 			Keyframes:     s.keyframes.Load(),
-			OutBytesTotal: stats[outBytesTotal].(uint64),
-			OutBytesAcked: stats[outBytesAcked].(uint64),
-			InBytesTotal:  stats[inBytesTotal].(uint64),
-			InBytesAcked:  stats[inBytesAcked].(uint64),
+			OutBytesTotal: tryUInt64(stats, outBytesTotal),
+			OutBytesAcked: tryUInt64(stats, outBytesAcked),
+			InBytesTotal:  tryUInt64(stats, inBytesTotal),
+			InBytesAcked:  tryUInt64(stats, inBytesAcked),
 		}, true
 	}
+
 	return nil, false
 }
+
+// sink stats sometimes returns strings instead of uint64
+func tryUInt64(stats map[string]interface{}, key string) uint64 {
+	switch val := stats[key].(type) {
+	case uint64:
+		return val
+	default:
+		logger.Infow(fmt.Sprintf("unexpected type for %s", key),
+			"type", fmt.Sprintf("%T", val),
+			"value", val,
+		)
+		return 0
+	}
+}