diff --git a/pkg/sip/inbound.go b/pkg/sip/inbound.go index fe5df0bd..323e3f22 100644 --- a/pkg/sip/inbound.go +++ b/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()), diff --git a/pkg/sip/inbound_test.go b/pkg/sip/inbound_test.go new file mode 100644 index 00000000..fe038ea5 --- /dev/null +++ b/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)) + }) + } +} diff --git a/pkg/sip/media_codecs.go b/pkg/sip/media_codecs.go index a22bf2a7..ae0aeb44 100644 --- a/pkg/sip/media_codecs.go +++ b/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 { diff --git a/pkg/sip/media_codecs_test.go b/pkg/sip/media_codecs_test.go new file mode 100644 index 00000000..baee3d14 --- /dev/null +++ b/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)) + }) + } +} diff --git a/pkg/sip/media_port.go b/pkg/sip/media_port.go index 0744d307..9d1fa461 100644 --- a/pkg/sip/media_port.go +++ b/pkg/sip/media_port.go @@ -743,6 +743,7 @@ 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} @@ -750,6 +751,15 @@ func (p *MediaPort) SetOffer(offerData []byte, codecs *msdk.CodecSet, enc sdp.En 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") diff --git a/pkg/sip/media_port_test.go b/pkg/sip/media_port_test.go index f514508e..64b693b4 100644 --- a/pkg/sip/media_port_test.go +++ b/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)) +} diff --git a/pkg/stats/monitor.go b/pkg/stats/monitor.go index 47679ffe..d016c1b8 100644 --- a/pkg/stats/monitor.go +++ b/pkg/stats/monitor.go @@ -75,6 +75,8 @@ type Monitor struct { durStage *prometheus.HistogramVec cpuLoad prometheus.Gauge sdpSize *prometheus.HistogramVec + sdpParsed *prometheus.CounterVec + codecOffered *prometheus.CounterVec nodeAvailable prometheus.GaugeFunc transfersTotal *prometheus.CounterVec transfersSucceeded *prometheus.CounterVec @@ -242,6 +244,22 @@ func (m *Monitor) Start(conf *config.Config) error { Buckets: sizeBuckets, }, []string{"type"})) + m.sdpParsed = mustRegister(m, prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "livekit", + Subsystem: "sip", + Name: "sdp_parsed_total", + Help: "Number of SDP bodies parsed successfully during SDP negotiation", + ConstLabels: prometheus.Labels{"node_id": conf.NodeID}, + }, []string{"dir", "provider"})) + + m.codecOffered = mustRegister(m, prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "livekit", + Subsystem: "sip", + Name: "codec_offered_total", + Help: "Number of SDP bodies that advertised a given audio codec", + ConstLabels: prometheus.Labels{"node_id": conf.NodeID}, + }, []string{"dir", "provider", "codec"})) + m.nodeAvailable = mustRegister(m, prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Namespace: "livekit", Subsystem: "sip", @@ -357,10 +375,28 @@ type CallMonitor struct { dir string fromHost string toHost string + provider atomic.Pointer[string] started atomic.Bool terminated atomic.Bool } +// ProviderUnknown is used when there is no provider information +const ProviderUnknown = "unknown" + +func (c *CallMonitor) SetProvider(provider string) { + if provider == "" { + return + } + c.provider.Store(&provider) +} + +func (c *CallMonitor) providerLabel() string { + if p := c.provider.Load(); p != nil { + return *p + } + return ProviderUnknown +} + func (c *CallMonitor) labelsShort(l prometheus.Labels) prometheus.Labels { out := prometheus.Labels{"dir": c.dir} for k, v := range l { @@ -488,6 +524,20 @@ func (c *CallMonitor) StageDurTimer(stage string) func() time.Duration { return prometheus.NewTimer(c.StageDur(stage)).ObserveDuration } +// PeerSDP increments SDP count and each individual codec from the SDP body. +// Should be called before codec selection such that failed negotiations are still counted +func (c *CallMonitor) PeerSDP(names []string) { + provider := c.providerLabel() + c.m.sdpParsed.With(prometheus.Labels{"dir": c.dir, "provider": provider}).Inc() + for _, name := range names { + c.m.codecOffered.With(prometheus.Labels{ + "dir": c.dir, + "provider": provider, + "codec": name, + }).Inc() + } +} + func (c *CallMonitor) SDPSize(sz int, isOffer bool) { typ := "answer" if isOffer {