TEL-764 Adding new metrics for SDP negotiation

pkg/sip/inbound.go

@@ -320,6 +320,23 @@ func sdpBodyFromRequest(req *sip.Request) []byte {
 	return req.Body()
 }
 
+func providerLabel(p *livekit.ProviderInfo) string {
+	switch p.GetType() {
+	case livekit.ProviderType_PROVIDER_TYPE_INTERNAL:
+		internalPrefix := "internal/"
+		if name := p.GetName(); name != "" {
+			return internalPrefix + strings.ToLower(name)
+		}
+
+		return internalPrefix + stats.ProviderUnknown
+	case livekit.ProviderType_PROVIDER_TYPE_EXTERNAL:
+		// External names are customer-supplied trunk names, left out to keep the label bounded.
+		return "external"
+	default:
+		return stats.ProviderUnknown
+	}
+}
+
 func updateRemoteFromSDP(media *MediaPort, log logger.Logger, codecs *msdk.CodecSet, body []byte) {
 	if len(body) == 0 || media == nil {
 		return
@@ -478,6 +495,7 @@ func (s *Server) processInvite(req *sip.Request, tx sip.ServerTransaction) (retE
 	if r.TrunkID != "" {
 		log = log.WithValues("sipTrunk", r.TrunkID)
 	}
+	cmon.SetProvider(providerLabel(r.ProviderInfo))
 
 	initial := &livekit.SIPCallInfo{
 		CallId:        string(cc.ID()),

pkg/sip/inbound_test.go

@@ -0,0 +1,72 @@
+// Copyright 2024 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// 	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sip
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/sip/pkg/stats"
+)
+
+func TestProviderLabel(t *testing.T) {
+	cases := []struct {
+		name string
+		info *livekit.ProviderInfo
+		exp  string
+	}{
+		{
+			name: "nil",
+			info: nil,
+			exp:  stats.ProviderUnknown,
+		},
+		{
+			name: "internal",
+			info: &livekit.ProviderInfo{Name: "someCarrier", Type: livekit.ProviderType_PROVIDER_TYPE_INTERNAL},
+			exp:  "internal/somecarrier",
+		},
+		{
+			name: "internal without a name",
+			info: &livekit.ProviderInfo{Type: livekit.ProviderType_PROVIDER_TYPE_INTERNAL},
+			exp:  "internal/unknown",
+		},
+		{
+			name: "external",
+			info: &livekit.ProviderInfo{
+				Id:   "ST_customerTrunk",
+				Name: "Some Customer's Twilio Trunk",
+				Type: livekit.ProviderType_PROVIDER_TYPE_EXTERNAL,
+			},
+			exp: "external",
+		},
+		{
+			name: "external without a name",
+			info: &livekit.ProviderInfo{Type: livekit.ProviderType_PROVIDER_TYPE_EXTERNAL},
+			exp:  "external",
+		},
+		{
+			name: "unknown type",
+			info: &livekit.ProviderInfo{Name: "someCarrier"},
+			exp:  stats.ProviderUnknown,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			require.Equal(t, c.exp, providerLabel(c.info))
+		})
+	}
+}

pkg/sip/media_codecs.go

@@ -18,6 +18,7 @@ package sip
 import (
 	"errors"
 	"fmt"
+	"slices"
 	"time"
 
 	_ "github.com/livekit/media-sdk/all"
@@ -43,6 +44,33 @@ func init() {
 	})
 }
 
+// Metric label used for advertised codecs that are not part of the internal
+// codec set, since their name is dropped during SDP parsing and to keep the
+// label bounded
+const codecOther = "other"
+
+func peerCodecNames(d sdp.MediaDesc) []string {
+	names := make([]string, 0, len(d.Codecs))
+	for _, c := range d.Codecs {
+		if d.DTMFType != 0 && c.Type == d.DTMFType {
+			// DTMF is parsed out of a=rtpmap into DTMFType, but its payload type is
+			// still listed in m=audio, where it resolves to no codec. Appended below.
+			continue
+		}
+		name := codecOther
+		if c.Codec != nil {
+			name = c.Codec.Info().SDPName
+		}
+		if !slices.Contains(names, name) {
+			names = append(names, name)
+		}
+	}
+	if d.DTMFType != 0 {
+		names = append(names, dtmf.SDPNameAndRate)
+	}
+	return names
+}
+
 func newMediaConfig(m *livekit.SIPMediaConfig, defaultTimeout time.Duration) (*sipMediaConfig, error) {
 	enc, err := sdpEncryption(m.Encryption)
 	if err != nil {

pkg/sip/media_codecs_test.go

@@ -0,0 +1,70 @@
+// Copyright 2024 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// 	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sip
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/livekit/media-sdk/sdp"
+)
+
+// sdpWithMedia builds a minimal SDP body with the given m= line and attributes.
+func sdpWithMedia(media string, attrs ...string) []byte {
+	body := "v=0\r\no=- 0 0 IN IP4 127.0.0.1\r\ns=-\r\nc=IN IP4 127.0.0.1\r\nt=0 0\r\n" + media + "\r\n"
+	for _, a := range attrs {
+		body += a + "\r\n"
+	}
+	return []byte(body)
+}
+
+// Testing edge cases that ParseOfferWith sometimes returns
+func TestPeerCodecNames(t *testing.T) {
+	cases := []struct {
+		name string
+		sdp  []byte
+		exp  []string
+	}{
+		{
+			// ParseMediaWith diverts telephone-event from a=rtpmap into DTMFType,
+			// but its payload type stays in m=audio and resolves to no codec.
+			// Without the skip it would be reported as an unsupported codec
+			name: "telephone-event is not an unsupported codec",
+			sdp: sdpWithMedia("m=audio 5004 RTP/AVP 0 101",
+				"a=rtpmap:0 PCMU/8000", "a=rtpmap:101 telephone-event/8000"),
+			exp: []string{"PCMU/8000", "telephone-event/8000"},
+		},
+		{
+			// No a=rtpmap at all, codecs resolved from the static payload types
+			name: "static payload types only",
+			sdp:  sdpWithMedia("m=audio 5004 RTP/AVP 0 8"),
+			exp:  []string{"PCMU/8000", "PCMA/8000"},
+		},
+		{
+			// A codec listed in both a=rtpmap and m=audio is parsed twice
+			name: "deduplicated",
+			sdp:  sdpWithMedia("m=audio 5004 RTP/AVP 0", "a=rtpmap:0 PCMU/8000"),
+			exp:  []string{"PCMU/8000"},
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			offer, err := sdp.ParseOfferWith(defaultCodecs, c.sdp)
+			require.NoError(t, err)
+			require.ElementsMatch(t, c.exp, peerCodecNames(offer.MediaDesc))
+		})
+	}
+}

pkg/sip/media_port.go

@@ -743,13 +743,23 @@ func (p *MediaPort) SetOffer(offerData []byte, codecs *msdk.CodecSet, enc sdp.En
 	if err != nil {
 		return nil, nil, SDPError{Err: err}
 	}
+	p.reportPeerCodecs(offer.MediaDesc)
 	answer, mc, err := offer.Answer(p.externalIP, p.Port(), enc)
 	if err != nil {
 		return nil, nil, SDPError{Err: err}
 	}
 	return answer, &MediaConf{MediaConfig: *mc}, nil
 }
 
+// Reported for inbound (SetOffer) only since outbound (SetAnswer) only contains the
+// codec picked by the end user, and not what they actually support
+func (p *MediaPort) reportPeerCodecs(d sdp.MediaDesc) {
+	if p.mon == nil {
+		return
+	}
+	p.mon.PeerSDP(peerCodecNames(d))
+}
+
 func (p *MediaPort) SetConfig(c *MediaConf) error {
 	if p.closed.IsBroken() {
 		return errors.New("media is already closed")

pkg/sip/media_port_test.go

@@ -30,6 +30,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 
 	msdk "github.com/livekit/media-sdk"
@@ -43,6 +44,11 @@ import (
 	"github.com/livekit/sip/pkg/stats"
 )
 
+const (
+	parsedMetric  = "livekit_sip_sdp_parsed_total"
+	offeredMetric = "livekit_sip_codec_offered_total"
+)
+
 func newTestCallMonitor(t testing.TB) *stats.CallMonitor {
 	mon, err := stats.NewMonitor(&config.Config{})
 	require.NoError(t, err)
@@ -51,6 +57,18 @@ func newTestCallMonitor(t testing.TB) *stats.CallMonitor {
 	return mon.NewCall(stats.Inbound, "test", "test")
 }
 
+func newTestMediaPort(t testing.TB, provider string) *MediaPort {
+	t.Helper()
+	mon := newTestCallMonitor(t)
+	mon.SetProvider(provider)
+	mp, err := NewMediaPortWith(1, logger.GetLogger(), mon, nil, &MediaOptions{
+		IP: netip.MustParseAddr("127.0.0.1"),
+	}, 8000)
+	require.NoError(t, err)
+	t.Cleanup(func() { mp.Close() })
+	return mp
+}
+
 type testUDPConn struct {
 	addr     netip.AddrPort
 	closed   chan struct{}
@@ -876,3 +894,90 @@ func TestMediaPortDTMF(t *testing.T) {
 		}
 	}
 }
+
+// Test util for incrementing prometheus counter metrics.
+func gatherCounter(t testing.TB, name string, labels map[string]string) float64 {
+	t.Helper()
+	families, err := prometheus.DefaultGatherer.Gather()
+	require.NoError(t, err)
+	var total float64
+	for _, f := range families {
+		// Matching metric
+		if f.GetName() != name {
+			continue
+		}
+	metrics:
+		for _, m := range f.GetMetric() {
+			got := make(map[string]string, len(m.GetLabel()))
+			for _, l := range m.GetLabel() {
+				got[l.GetName()] = l.GetValue()
+			}
+			// Matching labels
+			for k, v := range labels {
+				if got[k] != v {
+					continue metrics
+				}
+			}
+			total += m.GetCounter().GetValue()
+		}
+	}
+	return total
+}
+
+// Report codecs offered during SDP even when the offer fails to match any codecs
+func TestSetOfferReportsCodecsBeforeFailing(t *testing.T) {
+	mp := newTestMediaPort(t, "internal/somecarrier")
+
+	parsed := map[string]string{"dir": "in", "provider": "internal/somecarrier"}
+	other := map[string]string{"dir": "in", "provider": "internal/somecarrier", "codec": codecOther}
+	pcmu := map[string]string{"dir": "in", "provider": "internal/somecarrier", "codec": "PCMU/8000"}
+
+	parsedBefore := gatherCounter(t, parsedMetric, parsed)
+	otherBefore := gatherCounter(t, offeredMetric, other)
+	pcmuBefore := gatherCounter(t, offeredMetric, pcmu)
+
+	offer := sdpWithMedia("m=audio 5004 RTP/AVP 96", "a=rtpmap:96 SPEEX/16000")
+	_, _, err := mp.SetOffer(offer, defaultCodecs, sdp.EncryptionNone)
+	require.ErrorIs(t, err, sdp.ErrNoCommonMedia)
+
+	// Codecs that are not part of the internal set are classified as "other"
+	require.Equal(t, parsedBefore+1, gatherCounter(t, parsedMetric, parsed))
+	require.Equal(t, otherBefore+1, gatherCounter(t, offeredMetric, other))
+	require.Equal(t, pcmuBefore, gatherCounter(t, offeredMetric, pcmu))
+}
+
+func TestSetOfferReportsCodecsPerProvider(t *testing.T) {
+	mp := newTestMediaPort(t, "internal/somecarrier")
+
+	parsed := map[string]string{"dir": "in", "provider": "internal/somecarrier"}
+	pcmu := map[string]string{"dir": "in", "provider": "internal/somecarrier", "codec": "PCMU/8000"}
+	g722 := map[string]string{"dir": "in", "provider": "internal/somecarrier", "codec": "G722/8000"}
+
+	parsedBefore := gatherCounter(t, parsedMetric, parsed)
+	pcmuBefore := gatherCounter(t, offeredMetric, pcmu)
+	g722Before := gatherCounter(t, offeredMetric, g722)
+
+	offer := sdpWithMedia("m=audio 5004 RTP/AVP 0 9",
+		"a=rtpmap:0 PCMU/8000", "a=rtpmap:9 G722/8000")
+	_, _, err := mp.SetOffer(offer, defaultCodecs, sdp.EncryptionNone)
+	require.NoError(t, err)
+
+	require.Equal(t, parsedBefore+1, gatherCounter(t, parsedMetric, parsed))
+	require.Equal(t, pcmuBefore+1, gatherCounter(t, offeredMetric, pcmu))
+	require.Equal(t, g722Before+1, gatherCounter(t, offeredMetric, g722))
+}
+
+// Without SetProvider - the pre-auth path - offers still land somewhere rather
+// than being dropped.
+func TestSetOfferReportsUnknownProvider(t *testing.T) {
+	mp := newTestMediaPort(t, "")
+
+	parsed := map[string]string{"dir": "in", "provider": stats.ProviderUnknown}
+	before := gatherCounter(t, parsedMetric, parsed)
+
+	offer := sdpWithMedia("m=audio 5004 RTP/AVP 0", "a=rtpmap:0 PCMU/8000")
+	_, _, err := mp.SetOffer(offer, defaultCodecs, sdp.EncryptionNone)
+	require.NoError(t, err)
+
+	require.Equal(t, before+1, gatherCounter(t, parsedMetric, parsed))
+}