Dedup sourcePeerType (#4499)

Already stamped on every metric

Author: ilidemi

flow/connectors/mysql/cdc.go

@@ -1050,7 +1050,6 @@ func (c *MySqlConnector) recordColumnTypeChange(
 	eventType string,
 ) {
 	otelManager.Metrics.ColumnTypeChangesCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
-		attribute.String(otel_metrics.SourcePeerType, "mysql"),
 		attribute.String(otel_metrics.TypeChangeFromKey, string(from)),
 		attribute.String(otel_metrics.TypeChangeToKey, string(to)),
 		attribute.String(otel_metrics.SourceEventTypeKey, eventType),

flow/connectors/postgres/cdc.go

@@ -1306,7 +1306,6 @@ func processRelationMessage[Items model.Items](
 			p.logger.Warn(fmt.Sprintf("Detected column %s with type changed from %s to %s in table %s, but not propagating",
 				column.Name, prevRelMap[column.Name], currRelMap[column.Name], schemaDelta.SrcTableName))
 			p.otelManager.Metrics.ColumnTypeChangesCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
-				attribute.String(otel_metrics.SourcePeerType, "postgres"),
 				attribute.String(otel_metrics.TypeChangeFromKey, prevRelMap[column.Name]),
 				attribute.String(otel_metrics.TypeChangeToKey, currRelMap[column.Name]),
 				attribute.String(otel_metrics.SourceEventTypeKey, otel_metrics.SourceEventTypeEventMetadata),