Repository: atlassian/gostatsd
Pull request #652 (2024-01-09 05:46:22)

Add Lambda extension telemetry server
Currently the lambda extension binary will flush periodically based on the flush interval. Because the gostatsd runtime is suspended once the invocation is finished, the metric may:
- Not flush at all
- Flush at the next invocation

To address the flakiness of the flush when the lambda is not constantly invoked, we can subscribe to the telemetry API that AWS provides and listen for the `platform.runtimeDone` event to flush on within the extension.

This PR sets up the initial step to start a server and listen to lambda telemetry events, with a hook function defined on the above event.

A subsequent PR will add the logic to trigger a flush using the hook function.


Patch:
diff --git a/go.mod b/go.mod
index e83ec30..15beb9d 100644
--- a/go.mod
+++ b/go.mod
@@ -28,17 +28,11 @@ require (
 	stathat.com/c/consistent v1.0.0
 )
 
-require (
-	go.opentelemetry.io/proto/otlp v1.0.0
-	go.uber.org/multierr v1.9.0
-	golang.org/x/exp v0.0.0-20230905200255-921286631fa9
-)
-
 require (
 	github.com/PuerkitoBio/purell v1.1.1 // indirect
 	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
 	github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a // indirect
-	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/emicklei/go-restful/v3 v3.8.0 // indirect
@@ -53,7 +47,7 @@ require (
 	github.com/google/gnostic v0.5.7-v3refs // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/google/gofuzz v1.1.0 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
+	github.com/google/uuid v1.3.0 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/imdario/mergo v0.3.8 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
@@ -74,14 +68,13 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/yuin/gopher-lua v0.0.0-20210529063254-f4c35e4016d9 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
+	go.uber.org/multierr v1.9.0 // indirect
+	golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect
 	golang.org/x/oauth2 v0.12.0 // indirect
 	golang.org/x/sys v0.13.0 // indirect
 	golang.org/x/term v0.13.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20230913181813-007df8e322eb // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20230920204549-e6e6cdab5c13 // indirect
-	google.golang.org/grpc v1.58.2 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
diff --git a/go.sum b/go.sum
index bf188e4..680ca45 100644
--- a/go.sum
+++ b/go.sum
@@ -53,8 +53,8 @@ github.com/aws/aws-sdk-go v1.44.126/go.mod h1:y4AeaBuwd2Lk+GepC1E9v0qOiTws0MIWAX
 github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
 github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
-github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
-github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
@@ -103,7 +103,6 @@ github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
-github.com/golang/glog v1.1.0 h1:/d3pCKDPWNnvIWe0vVUpNP32qc8U3PDVxySP/y360qE=
 github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@@ -167,13 +166,12 @@ github.com/google/pprof v0.0.0-20201218002935-b9804c9f04c2/go.mod h1:kpwsk12EmLe
 github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/google-cloud-go-testing v0.0.0-20200911160855-bcd43fbb19e8/go.mod h1:dvDLG8qkwmyD9a/MJJN3XJcT3xFxOKAvTZGvuZmac9g=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
-github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 h1:YBftPWNWd4WwGqtY2yeZL2ef8rHAxPBD8KFhJpmcqms=
-github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0/go.mod h1:YN5jB8ie0yfIUg6VvR9Kz84aCaG7AsGZnLjhHbUqwPg=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
@@ -221,6 +219,7 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
@@ -283,8 +282,6 @@ go.opencensus.io v0.22.2/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
-go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lIVU/I=
-go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v80hjKIs5JXpM=
 go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
 go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/multierr v1.9.0 h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI=
@@ -566,11 +563,6 @@ google.golang.org/genproto v0.0.0-20201210142538-e3217bee35cc/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20201214200347-8c77b98c765d/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210108203827-ffc7fda8c3d7/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210226172003-ab064af71705/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
-google.golang.org/genproto v0.0.0-20230913181813-007df8e322eb h1:XFBgcDwm7irdHTbz4Zk2h7Mh+eis4nfJEFQFYzJzuIA=
-google.golang.org/genproto/googleapis/api v0.0.0-20230913181813-007df8e322eb h1:lK0oleSc7IQsUxO3U5TjL9DWlsxpEBemh+zpB7IqhWI=
-google.golang.org/genproto/googleapis/api v0.0.0-20230913181813-007df8e322eb/go.mod h1:KjSP20unUpOx5kyQUFa7k4OJg0qeJ7DEZflGDu2p6Bk=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20230920204549-e6e6cdab5c13 h1:N3bU/SQDCDyD6R528GJ/PwW9KjYcJA3dgyH+MovAkIM=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20230920204549-e6e6cdab5c13/go.mod h1:KSqppvjFjtoCI+KGd4PELB0qLNxdJHRGqRI09mB6pQA=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
 google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
 google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
@@ -587,8 +579,6 @@ google.golang.org/grpc v1.31.1/go.mod h1:N36X2cJ7JwdamYAgDz+s+rVMFjt3numwzf/HckM
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
 google.golang.org/grpc v1.34.0/go.mod h1:WotjhfgOW/POjDeRt8vscBtXq+2VjORFy659qA51WJ8=
 google.golang.org/grpc v1.35.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
-google.golang.org/grpc v1.58.2 h1:SXUpjxeVF3FKrTYQI4f4KvbGD5u2xccdYdurwowix5I=
-google.golang.org/grpc v1.58.2/go.mod h1:tgX3ZQDlNJGU96V6yHh1T/JeoBQ2TXdr43YbYSsCJk0=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
@@ -606,8 +596,8 @@ google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqw
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU=
 gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
diff --git a/internal/awslambda/extension/manager.go b/internal/awslambda/extension/manager.go
index 60b9550..3c966c2 100644
--- a/internal/awslambda/extension/manager.go
+++ b/internal/awslambda/extension/manager.go
@@ -18,6 +18,7 @@ import (
 	"go.uber.org/multierr"
 
 	"github.com/atlassian/gostatsd/internal/awslambda/extension/api"
+	"github.com/atlassian/gostatsd/internal/awslambda/extension/telemetry"
 )
 
 const (
@@ -26,9 +27,10 @@ const (
 )
 
 var (
-	ErrFailedRegistration = errors.New("extension failed to register with lambda")
-	ErrIssueProgress      = errors.New("unable to continue execution")
-	ErrServerEarlyExit    = errors.New("server has shutdown early without cause")
+	ErrFailedRegistration          = errors.New("extension failed to register with lambda")
+	ErrIssueProgress               = errors.New("unable continue execution")
+	ErrServerEarlyExit             = errors.New("server has shutdown early without cause")
+	ErrFailedTelemetrySubscription = errors.New("extension failed to subscribe to lambda telemetry API")
 )
 
 // Manager ensures that the lifetime management of the lambda
@@ -50,6 +52,8 @@ type manager struct {
 	domain       string
 	name         string
 	registeredID string
+
+	telemetryServer *telemetry.Server
 }
 
 var _ Manager = (*manager)(nil)
@@ -62,21 +66,33 @@ func NewManager(lambdaDomain string, lambdaFileName string, log logrus.FieldLogg
 		name:   lambdaFileName,
 	}
 
+	m.telemetryServer = telemetry.NewServer(telemetry.LambdaRuntimeAvailableAddr, log, telemetry.NoopHook())
+
 	return m
 }
 
 func (m *manager) Run(parent context.Context, server Server) error {
+	var wg wait.Group
+	var chErrs = make(chan error, 3)
+
 	// Init Phase of the lambda
 	if err := m.register(parent); err != nil {
 		return err
 	}
 
+	// Start telemetry server
 	ctx, cancel := context.WithCancel(parent)
-	// Start gostatsd server
-	var wg wait.Group
-	var chErrs = make(chan error, 2)
 	defer cancel()
+	wg.StartWithContext(ctx, func(c context.Context) {
+		chErrs <- m.telemetryServer.Start(c)
+	})
+
+	if err := m.subscribeToTelemetry(ctx); err != nil {
+		m.log.WithError(err).Error("error subscribing to lambda telemetry endpoint")
+		return err
+	}
 
+	// Start gostatsd server
 	wg.StartWithContext(ctx, func(c context.Context) {
 		err := server.Run(c)
 		// Shutting down due to context is an acceptable
@@ -214,6 +230,47 @@ func (m *manager) register(ctx context.Context) error {
 	return nil
 }
 
+func (m *manager) subscribeToTelemetry(ctx context.Context) error {
+	b, err := jsoniter.Marshal(telemetry.SubscriptionRequest{
+		SchemaVersion: telemetry.ApiVersion,
+		Types:         []string{telemetry.PlatformSubscriptionType},
+		Destination:   &telemetry.SubscriptionDestination{URI: m.telemetryServer.Endpoint(), Protocol: telemetry.ProtocolHTTP},
+		// we buffer for the minimum time on the lambda side to avoid adding additional billedDuration to the lambda
+		Buffering: &telemetry.SubscriptionBufferingConfig{TimeoutMs: telemetry.MinBufferingTimeoutMs},
+	})
+	if err != nil {
+		return err
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, telemetry.SubscribeEndpoint.GetUrl(m.domain), bytes.NewReader(b))
+	if err != nil {
+		return err
+	}
+	req.Header.Set(api.LambdaExtensionIdentifierHeaderKey, m.registeredID)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := m.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+	}()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		m.log.WithError(err).WithFields(map[string]interface{}{
+			"statusCode": resp.StatusCode,
+			"msg":        string(body),
+		}).Error("Error subscribing to telemetry API")
+		return ErrFailedTelemetrySubscription
+	}
+
+	m.log.Info("Successfully subscribed to telemetry API")
+
+	return nil
+}
+
 func (m *manager) nextEvent(ctx context.Context) (*api.EventNextPayload, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, api.EventEndpoint.GetUrl(m.domain), http.NoBody)
 	if err != nil {
@@ -226,10 +283,7 @@ func (m *manager) nextEvent(ctx context.Context) (*api.EventNextPayload, error)
 	if err != nil {
 		return nil, err
 	}
-	defer func() {
-		_, _ = io.Copy(io.Discard, resp.Body)
-		_ = resp.Body.Close()
-	}()
+	defer resp.Body.Close()
 
 	switch resp.StatusCode {
 	case http.StatusOK:
diff --git a/internal/awslambda/extension/telemetry/api.go b/internal/awslambda/extension/telemetry/api.go
new file mode 100644
index 0000000..fc31aa6
--- /dev/null
+++ b/internal/awslambda/extension/telemetry/api.go
@@ -0,0 +1,44 @@
+package telemetry
+
+import "github.com/atlassian/gostatsd/internal/awslambda/extension/api"
+
+const (
+	ApiVersion                              = "2022-07-01"
+	SubscribeEndpoint        api.LambdaPath = "/" + ApiVersion + "/telemetry"
+	ProtocolHTTP             string         = "HTTP"
+	PlatformSubscriptionType string         = "platform"
+	MinBufferingTimeoutMs    int            = 25
+
+	RuntimeDone EventType = "platform.runtimeDone"
+	// LambdaRuntimeAvailableAddr uses the sandbox hostname must be used in the lambda runtime
+	// note the port component can be configured
+	LambdaRuntimeAvailableAddr = "sandbox:8083"
+)
+
+type EventType string
+
+// Event is the telemetry event that the lambda server pushes.
+//
+// Full schema can be found here: https://docs.aws.amazon.com/lambda/latest/dg/telemetry-schema-reference.html
+type Event struct {
+	// Type is the only field we deserialise as it is used for flushing metrics
+	Type EventType `json:"type"`
+}
+
+type SubscriptionRequest struct {
+	SchemaVersion string                       `json:"schemaVersion"`
+	Types         []string                     `json:"types"`
+	Buffering     *SubscriptionBufferingConfig `json:"buffering,omitempty"`
+	Destination   *SubscriptionDestination     `json:"destination,omitempty"`
+}
+
+type SubscriptionDestination struct {
+	Protocol string `json:"protocol,omitempty"`
+	URI      string `json:"URI"`
+}
+
+type SubscriptionBufferingConfig struct {
+	MaxItems  int         `json:"maxItems,omitempty"`
+	MaxBytes  interface{} `json:"maxBytes,omitempty"`
+	TimeoutMs int         `json:"timeoutMs,omitempty"`
+}
diff --git a/internal/awslambda/extension/telemetry/doc.go b/internal/awslambda/extension/telemetry/doc.go
new file mode 100644
index 0000000..77e53cb
--- /dev/null
+++ b/internal/awslambda/extension/telemetry/doc.go
@@ -0,0 +1,8 @@
+// Package telemetry defines the apis and server required
+// to interact with the subscribe to the Telemetry API for AWS lambdas.
+// This is to enable the trigger of specific actions such as metric flushes on
+// certain runtime events.
+//
+// More information on the Telemetry API can be found at :
+// https://docs.aws.amazon.com/lambda/latest/dg/telemetry-api.html
+package telemetry
diff --git a/internal/awslambda/extension/telemetry/server.go b/internal/awslambda/extension/telemetry/server.go
new file mode 100644
index 0000000..5dc149e
--- /dev/null
+++ b/internal/awslambda/extension/telemetry/server.go
@@ -0,0 +1,85 @@
+package telemetry
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/mux"
+	jsoniter "github.com/json-iterator/go"
+	"github.com/sirupsen/logrus"
+)
+
+type RuntimeDoneHook func()
+
+func NoopHook() RuntimeDoneHook {
+	return func() {}
+}
+
+type Server struct {
+	addr       string
+	log        logrus.FieldLogger
+	f          RuntimeDoneHook
+	httpServer *http.Server
+}
+
+func NewServer(addr string, log logrus.FieldLogger, hook RuntimeDoneHook) *Server {
+	ts := &Server{
+		log: log,
+		f:   hook,
+	}
+
+	mx := mux.NewRouter()
+	mx.HandleFunc("/telemetry", ts.eventHandler).Methods(http.MethodPost)
+
+	ts.httpServer = &http.Server{Addr: addr, Handler: mx}
+
+	return ts
+}
+
+func (s *Server) Start(ctx context.Context) error {
+	go func() {
+		<-ctx.Done()
+		c, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel()
+		err := s.httpServer.Shutdown(c)
+		if err != nil {
+			s.log.WithError(err).Info("Did not shutdown gracefully")
+		}
+	}()
+
+	if err := s.httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		s.log.WithError(err).Error("Server error")
+		return err
+	}
+
+	s.log.Info("Shutdown telemetry server")
+
+	return nil
+}
+
+func (s *Server) eventHandler(_ http.ResponseWriter, r *http.Request) {
+	b, err := io.ReadAll(r.Body)
+	if err != nil {
+		s.log.WithError(err).Error("Error reading body")
+	}
+
+	var telePayload []Event
+	err = jsoniter.Unmarshal(b, &telePayload)
+	if err != nil {
+		s.log.WithError(err).Error("Error unmarshaling telemetry request")
+	}
+
+	for _, p := range telePayload {
+		if p.Type == RuntimeDone {
+			s.f()
+		}
+	}
+}
+
+func (s *Server) Endpoint() string {
+	return fmt.Sprintf("http://%s/telemetry", s.httpServer.Addr)
+}
diff --git a/pkg/backends/otlp/client.go b/pkg/backends/otlp/client.go
deleted file mode 100644
index 2c4d1e2..0000000
--- a/pkg/backends/otlp/client.go
+++ /dev/null
@@ -1,30 +0,0 @@
-package otlp
-
-import (
-	"context"
-
-	"github.com/atlassian/gostatsd"
-)
-
-const (
-	namedBackend = `otlp`
-)
-
-// Client contains additional meta data in order
-// to export values as OTLP metrics.
-// The zero value is not safe to use.
-type Client struct {
-	resources []string
-}
-
-var _ gostatsd.Backend = (*Client)(nil)
-
-func (Client) Name() string { return namedBackend }
-
-func (Client) SendEvent(ctx context.Context, e *gostatsd.Event) error {
-	return nil
-}
-
-func (Client) SendMetricsAsync(ctx context.Context, mm *gostatsd.MetricMap, cb gostatsd.SendCallback) {
-	cb(nil)
-}
diff --git a/pkg/backends/otlp/internal/data/gauge.go b/pkg/backends/otlp/internal/data/gauge.go
deleted file mode 100644
index f597bec..0000000
--- a/pkg/backends/otlp/internal/data/gauge.go
+++ /dev/null
@@ -1,23 +0,0 @@
-package data
-
-import (
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-type Gauge struct {
-	raw *v1metrics.Gauge
-}
-
-func NewGauge(datapoints ...NumberDataPoint) Gauge {
-	g := Gauge{
-		raw: &v1metrics.Gauge{
-			DataPoints: make([]*v1metrics.NumberDataPoint, 0, len(datapoints)),
-		},
-	}
-
-	for i := 0; i < len(datapoints); i++ {
-		g.raw.DataPoints = append(g.raw.DataPoints, datapoints[i].raw)
-	}
-
-	return g
-}
diff --git a/pkg/backends/otlp/internal/data/histograms.go b/pkg/backends/otlp/internal/data/histograms.go
deleted file mode 100644
index 07fe2b5..0000000
--- a/pkg/backends/otlp/internal/data/histograms.go
+++ /dev/null
@@ -1,87 +0,0 @@
-package data
-
-import (
-	"math"
-	"slices"
-
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-	"golang.org/x/exp/maps"
-)
-
-type Histogram struct {
-	raw *v1metrics.Histogram
-}
-
-type HistogramDataPoint struct {
-	raw *v1metrics.HistogramDataPoint
-}
-
-func NewHistogram(datapoints ...HistogramDataPoint) Histogram {
-	ht := Histogram{
-		raw: &v1metrics.Histogram{
-			AggregationTemporality: v1metrics.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA,
-			DataPoints:             make([]*v1metrics.HistogramDataPoint, 0, len(datapoints)),
-		},
-	}
-
-	for i := 0; i < len(datapoints); i++ {
-		ht.raw.DataPoints = append(
-			ht.raw.DataPoints,
-			datapoints[i].raw,
-		)
-	}
-
-	return ht
-}
-
-func WithHistogramDataPointAttributes(attrs Map) func(HistogramDataPoint) {
-	return func(hdp HistogramDataPoint) {
-		hdp.raw.Attributes = attrs.unwrap()
-	}
-}
-
-func WithHistogramDataPointStatistics(values []float64) func(HistogramDataPoint) {
-	return func(hdp HistogramDataPoint) {
-		hdp.raw.Sum = new(float64)
-		hdp.raw.Min = &values[0]
-		hdp.raw.Max = &values[len(values)-1]
-		hdp.raw.Count = uint64(len(values))
-
-		for _, v := range values {
-			*hdp.raw.Sum += v
-			*hdp.raw.Min = math.Min(*hdp.raw.Min, v)
-			*hdp.raw.Max = math.Max(*hdp.raw.Max, v)
-		}
-	}
-}
-
-func WithHistogramDataPointBucketValues[Buckets ~map[float64]uint64](buckets Buckets) func(HistogramDataPoint) {
-	return func(hdp HistogramDataPoint) {
-		bounds := maps.Keys(buckets)
-		slices.Sort(bounds)
-
-		hdp.raw.BucketCounts = make([]uint64, len(buckets))
-		hdp.raw.ExplicitBounds = make([]float64, len(buckets)-1)
-
-		for i, bound := range bounds {
-			hdp.raw.BucketCounts[i] = buckets[bound]
-			if !math.IsInf(bound, 1) {
-				hdp.raw.ExplicitBounds[i] = bound
-			}
-		}
-	}
-}
-
-func NewHistogramDataPoint(timestamp uint64, opts ...func(HistogramDataPoint)) HistogramDataPoint {
-	dp := HistogramDataPoint{
-		raw: &v1metrics.HistogramDataPoint{
-			TimeUnixNano: timestamp,
-		},
-	}
-
-	for i := 0; i < len(opts); i++ {
-		opts[i](dp)
-	}
-
-	return dp
-}
diff --git a/pkg/backends/otlp/internal/data/instrumentation_scope.go b/pkg/backends/otlp/internal/data/instrumentation_scope.go
deleted file mode 100644
index bc728a2..0000000
--- a/pkg/backends/otlp/internal/data/instrumentation_scope.go
+++ /dev/null
@@ -1,35 +0,0 @@
-package data
-
-import v1common "go.opentelemetry.io/proto/otlp/common/v1"
-
-type InstrumentationScope struct {
-	raw *v1common.InstrumentationScope
-}
-
-func WithInstrumentationScopeAttributes(attributes Map) func(InstrumentationScope) {
-	return func(is InstrumentationScope) {
-		is.raw.Attributes = attributes.unwrap()
-	}
-}
-
-// NewInstrumentationScope is used for denote how the data was generated,
-// this will always be set to `gostatsd` as the name, and the deployed version.
-// Additional attributes can be set that could be used for filtering later on.
-func NewInstrumentationScope(name string, version string, opts ...func(InstrumentationScope)) InstrumentationScope {
-	is := InstrumentationScope{
-		raw: &v1common.InstrumentationScope{
-			Name:    name,
-			Version: version,
-		},
-	}
-
-	for _, opt := range opts {
-		opt(is)
-	}
-
-	return is
-}
-
-func (is InstrumentationScope) Attributes() Map {
-	return Map{raw: &is.raw.Attributes}
-}
diff --git a/pkg/backends/otlp/internal/data/map.go b/pkg/backends/otlp/internal/data/map.go
deleted file mode 100644
index fba90b3..0000000
--- a/pkg/backends/otlp/internal/data/map.go
+++ /dev/null
@@ -1,108 +0,0 @@
-package data
-
-import (
-	"slices"
-	"strings"
-
-	v1common "go.opentelemetry.io/proto/otlp/common/v1"
-	"google.golang.org/protobuf/proto"
-)
-
-type Map struct {
-	raw *[]*v1common.KeyValue
-}
-
-func NewMap(opts ...func(Map)) Map {
-	m := Map{
-		raw: new([]*v1common.KeyValue),
-	}
-	for _, opt := range opts {
-		opt(m)
-	}
-	return m
-}
-
-// WithDelimtedStrings splits each value based on the delim passed,
-// if there is no delim present, then the entire string is added to the key
-// and uses an empty string as the value
-func WithDelimitedStrings[Tags ~[]string](delim string, tags Tags) func(m Map) {
-	return func(m Map) {
-		for _, kv := range tags {
-			idx := strings.Index(kv, delim)
-			switch idx {
-			case -1:
-				// No delimiter found
-				m.Insert(kv, "")
-			default:
-				m.Insert(kv[:idx], kv[idx+1:])
-			}
-		}
-	}
-}
-
-func WithStatsdDelimitedTags[Tags ~[]string](tags Tags) func(Map) {
-	return WithDelimitedStrings(":", tags)
-}
-
-func (m Map) Equal(mm Map) bool {
-	return proto.Equal(
-		&v1common.KeyValueList{Values: *m.raw},
-		&v1common.KeyValueList{Values: *mm.raw},
-	)
-}
-
-func (m Map) Insert(key string, value string) {
-	index := slices.IndexFunc(*m.raw, func(kv *v1common.KeyValue) bool {
-		return kv.Key == key
-	})
-
-	val := &v1common.AnyValue{
-		Value: &v1common.AnyValue_StringValue{
-			StringValue: value,
-		},
-	}
-
-	switch index {
-	case -1:
-		// The value doesn't exist within the map, appending to end
-		*m.raw = append(*m.raw, &v1common.KeyValue{
-			Key:   key,
-			Value: val,
-		})
-	default:
-		if v, ok := (*m.raw)[index].Value.Value.(*v1common.AnyValue_StringValue); ok && v.StringValue == value {
-			return
-		}
-
-		v, ok := (*m.raw)[index].Value.Value.(*v1common.AnyValue_ArrayValue)
-		if !ok {
-			// Convert a simple key value into an array of values
-			v = &v1common.AnyValue_ArrayValue{
-				ArrayValue: &v1common.ArrayValue{
-					Values: []*v1common.AnyValue{
-						{Value: v},
-					},
-				},
-			}
-		}
-		v.ArrayValue.Values = append(v.ArrayValue.Values, val)
-	}
-}
-
-func (m Map) Merge(mm Map) {
-	for _, kv := range *mm.raw {
-		switch v := kv.Value.Value.(type) {
-		case *v1common.AnyValue_StringValue:
-			m.Insert(kv.Key, v.StringValue)
-
-		case *v1common.AnyValue_ArrayValue:
-			for _, vv := range v.ArrayValue.Values {
-				m.Insert(kv.Key, vv.GetStringValue())
-			}
-		}
-	}
-}
-
-func (m Map) unwrap() []*v1common.KeyValue {
-	return *m.raw
-}
diff --git a/pkg/backends/otlp/internal/data/metric.go b/pkg/backends/otlp/internal/data/metric.go
deleted file mode 100644
index fd81de4..0000000
--- a/pkg/backends/otlp/internal/data/metric.go
+++ /dev/null
@@ -1,40 +0,0 @@
-package data
-
-import (
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-type Metric struct {
-	raw *v1metrics.Metric
-}
-
-// NewMetric creates a new metric wrapper and
-// sets the metric value based on the provided function.
-func NewMetric(name string) Metric {
-	return Metric{
-		raw: &v1metrics.Metric{
-			Name: name,
-		},
-	}
-}
-
-func (m Metric) SetGauge(g Gauge) Metric {
-	m.raw.Data = &v1metrics.Metric_Gauge{
-		Gauge: g.raw,
-	}
-	return m
-}
-
-func (m Metric) SetSum(s Sum) Metric {
-	m.raw.Data = &v1metrics.Metric_Sum{
-		Sum: s.raw,
-	}
-	return m
-}
-
-func (m Metric) SetHistogram(h Histogram) Metric {
-	m.raw.Data = &v1metrics.Metric_Histogram{
-		Histogram: h.raw,
-	}
-	return m
-}
diff --git a/pkg/backends/otlp/internal/data/metric_request.go b/pkg/backends/otlp/internal/data/metric_request.go
deleted file mode 100644
index d263904..0000000
--- a/pkg/backends/otlp/internal/data/metric_request.go
+++ /dev/null
@@ -1,50 +0,0 @@
-package data
-
-import (
-	"bytes"
-	"context"
-	"net/http"
-
-	v1export "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-	"google.golang.org/protobuf/proto"
-)
-
-const (
-	metricRequestContentType = "application/x-protobuf"
-)
-
-type metricsRequest struct {
-	raw *v1export.ExportMetricsServiceRequest
-}
-
-func NewMetricsRequest(ctx context.Context, endpoint string, metrics ...ResourceMetrics) (*http.Request, error) {
-	mr := metricsRequest{
-		raw: &v1export.ExportMetricsServiceRequest{
-			ResourceMetrics: make([]*v1metrics.ResourceMetrics, 0, len(metrics)),
-		},
-	}
-
-	for i := 0; i < len(metrics); i++ {
-		mr.raw.ResourceMetrics = append(mr.raw.ResourceMetrics, metrics[i].raw)
-	}
-
-	buf, err := proto.Marshal(mr.raw)
-	if err != nil {
-		return nil, err
-	}
-
-	req, err := http.NewRequestWithContext(
-		ctx,
-		http.MethodPost,
-		endpoint,
-		bytes.NewBuffer(buf),
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	req.Header.Set("Content-Type", metricRequestContentType)
-
-	return req, nil
-}
diff --git a/pkg/backends/otlp/internal/data/metric_response.go b/pkg/backends/otlp/internal/data/metric_response.go
deleted file mode 100644
index c88e54c..0000000
--- a/pkg/backends/otlp/internal/data/metric_response.go
+++ /dev/null
@@ -1,33 +0,0 @@
-package data
-
-import (
-	"fmt"
-	"io"
-	"net/http"
-
-	v1export "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
-	"go.uber.org/multierr"
-	"google.golang.org/protobuf/proto"
-)
-
-func ProcessMetricResponse(resp *http.Response) (dropped int64, errs error) {
-	buf, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return 0, err
-	}
-
-	var response v1export.ExportMetricsServiceResponse
-	if err := proto.Unmarshal(buf, &response); err != nil {
-		return 0, err
-	}
-
-	if ps := response.PartialSuccess; ps != nil && ps.ErrorMessage != "" {
-		if ps.RejectedDataPoints > 0 {
-			dropped = ps.RejectedDataPoints
-			errs = multierr.Append(errs, fmt.Errorf("dataloss: dropped %d metrics", ps.RejectedDataPoints))
-		}
-		errs = multierr.Append(errs, fmt.Errorf("failed to send metrics: %s", ps.ErrorMessage))
-	}
-
-	return dropped, errs
-}
diff --git a/pkg/backends/otlp/internal/data/number_datapoint.go b/pkg/backends/otlp/internal/data/number_datapoint.go
deleted file mode 100644
index 6ca5352..0000000
--- a/pkg/backends/otlp/internal/data/number_datapoint.go
+++ /dev/null
@@ -1,46 +0,0 @@
-package data
-
-import (
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-type NumberDataPoint struct {
-	raw *v1metrics.NumberDataPoint
-}
-
-func WithNumberDataPointDelimtedTags[Tags ~[]string](tags Tags) func(NumberDataPoint) {
-	return func(ndp NumberDataPoint) {
-		ndp.raw.Attributes = NewMap(WithStatsdDelimitedTags(tags)).unwrap()
-	}
-}
-
-func WithNumberDatapointIntValue(value int64) func(NumberDataPoint) {
-	return func(ndp NumberDataPoint) {
-		ndp.raw.Value = &v1metrics.NumberDataPoint_AsInt{
-			AsInt: value,
-		}
-	}
-}
-
-func WithNumberDataPointDoubleValue(value float64) func(NumberDataPoint) {
-	return func(ndp NumberDataPoint) {
-		ndp.raw.Value = &v1metrics.NumberDataPoint_AsDouble{
-			AsDouble: value,
-		}
-	}
-}
-
-// NewNumberDataPoint applies options in order so the last value set wins
-func NewNumberDataPoint(timestamp uint64, opts ...func(NumberDataPoint)) NumberDataPoint {
-	dp := NumberDataPoint{
-		raw: &v1metrics.NumberDataPoint{
-			TimeUnixNano: timestamp,
-		},
-	}
-
-	for _, opt := range opts {
-		opt(dp)
-	}
-
-	return dp
-}
diff --git a/pkg/backends/otlp/internal/data/resource.go b/pkg/backends/otlp/internal/data/resource.go
deleted file mode 100644
index 20cd5ff..0000000
--- a/pkg/backends/otlp/internal/data/resource.go
+++ /dev/null
@@ -1,29 +0,0 @@
-package data
-
-import (
-	v1resource "go.opentelemetry.io/proto/otlp/resource/v1"
-)
-
-type Resource struct {
-	raw *v1resource.Resource
-}
-
-func WithResourceAttributes(opts ...func(Map)) func(Resource) {
-	return func(r Resource) {
-		r.Attributes().Merge(NewMap(opts...))
-	}
-}
-
-func NewResource(opts ...func(Resource)) Resource {
-	r := Resource{
-		raw: &v1resource.Resource{},
-	}
-	for _, opt := range opts {
-		opt(r)
-	}
-	return r
-}
-
-func (r Resource) Attributes() Map {
-	return Map{raw: &(*r.raw).Attributes}
-}
diff --git a/pkg/backends/otlp/internal/data/resource_metrics.go b/pkg/backends/otlp/internal/data/resource_metrics.go
deleted file mode 100644
index 7add917..0000000
--- a/pkg/backends/otlp/internal/data/resource_metrics.go
+++ /dev/null
@@ -1,22 +0,0 @@
-package data
-
-import v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-
-type ResourceMetrics struct {
-	raw *v1metrics.ResourceMetrics
-}
-
-func NewResourceMetrics(resource Resource, scopeMetrics ...ScopeMetrics) ResourceMetrics {
-	rm := ResourceMetrics{
-		raw: &v1metrics.ResourceMetrics{
-			Resource:     resource.raw,
-			ScopeMetrics: make([]*v1metrics.ScopeMetrics, 0, len(scopeMetrics)),
-		},
-	}
-
-	for i := 0; i < len(scopeMetrics); i++ {
-		rm.raw.ScopeMetrics = append(rm.raw.ScopeMetrics, scopeMetrics[i].raw)
-	}
-
-	return rm
-}
diff --git a/pkg/backends/otlp/internal/data/scope_metrics.go b/pkg/backends/otlp/internal/data/scope_metrics.go
deleted file mode 100644
index eee67f8..0000000
--- a/pkg/backends/otlp/internal/data/scope_metrics.go
+++ /dev/null
@@ -1,24 +0,0 @@
-package data
-
-import (
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-type ScopeMetrics struct {
-	raw *v1metrics.ScopeMetrics
-}
-
-func NewScopeMetrics(is InstrumentationScope, metrics ...Metric) ScopeMetrics {
-	sm := ScopeMetrics{
-		raw: &v1metrics.ScopeMetrics{
-			Scope:   is.raw,
-			Metrics: make([]*v1metrics.Metric, 0, len(metrics)),
-		},
-	}
-
-	for i := 0; i < len(metrics); i++ {
-		sm.raw.Metrics = append(sm.raw.Metrics, metrics[i].raw)
-	}
-
-	return sm
-}
diff --git a/pkg/backends/otlp/internal/data/sum.go b/pkg/backends/otlp/internal/data/sum.go
deleted file mode 100644
index 8eb0d87..0000000
--- a/pkg/backends/otlp/internal/data/sum.go
+++ /dev/null
@@ -1,25 +0,0 @@
-package data
-
-import (
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-type Sum struct {
-	raw *v1metrics.Sum
-}
-
-func NewSum(datapoints ...NumberDataPoint) Sum {
-	s := Sum{
-		raw: &v1metrics.Sum{
-			IsMonotonic:            false,
-			AggregationTemporality: v1metrics.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA,
-			DataPoints:             make([]*v1metrics.NumberDataPoint, 0, len(datapoints)),
-		},
-	}
-
-	for i := 0; i < len(datapoints); i++ {
-		s.raw.DataPoints = append(s.raw.DataPoints, datapoints[i].raw)
-	}
-
-	return s
-}
diff --git a/pkg/backends/statsdaemon/tls.go b/pkg/backends/statsdaemon/tls.go
index 2419ae2..6f93588 100644
--- a/pkg/backends/statsdaemon/tls.go
+++ b/pkg/backends/statsdaemon/tls.go
@@ -4,7 +4,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
-	"os"
+	"io/ioutil"
 )
 
 func getTLSConfiguration(caPath, certPath, keyPath string, enable bool) (*tls.Config, error) {
@@ -20,7 +20,7 @@ func getTLSConfiguration(caPath, certPath, keyPath string, enable bool) (*tls.Co
 	}
 
 	if caPath != "" {
-		caPEM, err := os.ReadFile(caPath)
+		caPEM, err := ioutil.ReadFile(caPath)
 		if err != nil {
 			return nil, fmt.Errorf("[%s] error reading TLS CA: %v", BackendName, err)
 		}


Test patch:
diff --git a/internal/awslambda/extension/manager_test.go b/internal/awslambda/extension/manager_test.go
index da7869b..94d73c2 100644
--- a/internal/awslambda/extension/manager_test.go
+++ b/internal/awslambda/extension/manager_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -17,7 +18,7 @@ import (
 	"github.com/tilinna/clock"
 
 	"github.com/atlassian/gostatsd/internal/awslambda/extension/api"
-	"github.com/atlassian/gostatsd/internal/fixtures"
+	"github.com/atlassian/gostatsd/internal/awslambda/extension/telemetry"
 	"github.com/atlassian/gostatsd/pkg/fakesocket"
 )
 
@@ -126,7 +127,7 @@ func ExitErrorHandler(tb testing.TB, statusCode int) http.Handler {
 	})
 }
 
-func EventNextHandler(tb testing.TB, ctx context.Context, statusCode, shutdownAfter int) http.Handler {
+func EventNextHandler(tb testing.TB, statusCode, shutdownAfter int) http.Handler {
 	count := 0
 	return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
 		// Enforcing the correct method is used
@@ -138,7 +139,6 @@ func EventNextHandler(tb testing.TB, ctx context.Context, statusCode, shutdownAf
 		case http.StatusOK:
 			payload := &api.EventNextPayload{
 				EventType:          api.Invoke,
-				Deadline:           clock.FromContext(ctx).Now().UTC().Add(100 * time.Millisecond).UnixMilli(),
 				RequestID:          fmt.Sprint(count),
 				InvokedFunctionARN: "local:dev:gostatsd-extension-manager",
 			}
@@ -162,6 +162,21 @@ func EventNextHandler(tb testing.TB, ctx context.Context, statusCode, shutdownAf
 	})
 }
 
+func TelemetryHandler(tb testing.TB, statusCode int) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		require.Equal(tb, http.MethodPut, r.Method)
+		require.Contains(tb, r.Header, api.LambdaExtensionIdentifierHeaderKey)
+		require.Equal(tb, tb.Name(), r.Header.Get(api.LambdaExtensionIdentifierHeaderKey))
+		w.WriteHeader(statusCode)
+	})
+}
+
+func availableAddr() string {
+	l, _ := net.Listen("tcp", ":0")
+	defer l.Close()
+	return l.Addr().String()
+}
+
 func TestManagerRegister(t *testing.T) {
 	t.Parallel()
 
@@ -227,53 +242,68 @@ func TestManagerDo(t *testing.T) {
 	t.Parallel()
 
 	testCases := []struct {
-		Scenario      string
-		EventStatus   int
-		ShutdownAfter int
-		MockDelay     time.Duration
-		MockError     error
+		Scenario        string
+		EventStatus     int
+		TelemetryStatus int
+		ShutdownAfter   int
+		MockDelay       time.Duration
+		MockError       error
 
 		ExpectError error
 	}{
 		{
-			Scenario:      "Normal operation",
-			EventStatus:   http.StatusOK,
-			ShutdownAfter: 3,
-			MockDelay:     time.Second,
-			MockError:     nil,
-			ExpectError:   nil,
+			Scenario:        "Normal operation",
+			EventStatus:     http.StatusOK,
+			TelemetryStatus: http.StatusOK,
+			ShutdownAfter:   3,
+			MockDelay:       time.Second,
+			MockError:       nil,
+			ExpectError:     nil,
 		},
 		{
-			Scenario:      "Operational failure sending heartbeart",
-			EventStatus:   http.StatusInternalServerError,
-			ShutdownAfter: 2,
-			MockDelay:     time.Second,
-			MockError:     nil,
-			ExpectError:   ErrIssueProgress,
+			Scenario:        "Operational failure sending heartbeart",
+			EventStatus:     http.StatusInternalServerError,
+			TelemetryStatus: http.StatusOK,
+			ShutdownAfter:   2,
+			MockDelay:       time.Second,
+			MockError:       nil,
+			ExpectError:     ErrIssueProgress,
 		},
 		{
-			Scenario:      "Server has shutdown early without cause",
-			EventStatus:   http.StatusOK,
-			ShutdownAfter: 3,
-			MockDelay:     0,
-			MockError:     nil,
-			ExpectError:   ErrServerEarlyExit,
+			Scenario:        "Server has shutdown early without cause",
+			EventStatus:     http.StatusOK,
+			TelemetryStatus: http.StatusOK,
+			ShutdownAfter:   3,
+			MockDelay:       0,
+			MockError:       nil,
+			ExpectError:     ErrServerEarlyExit,
 		},
 		{
-			Scenario:      "Server configuration was wrong resulting in during init",
-			EventStatus:   http.StatusOK,
-			ShutdownAfter: 0,
-			MockDelay:     0,
-			MockError:     fakesocket.ErrClosedConnection,
-			ExpectError:   fakesocket.ErrClosedConnection,
+			Scenario:        "Server configuration was wrong resulting in during init",
+			EventStatus:     http.StatusOK,
+			TelemetryStatus: http.StatusOK,
+			ShutdownAfter:   0,
+			MockDelay:       0,
+			MockError:       fakesocket.ErrClosedConnection,
+			ExpectError:     fakesocket.ErrClosedConnection,
 		},
 		{
-			Scenario:      "Server failed throughout runtime and successfully committed to lambda",
-			EventStatus:   http.StatusOK,
-			ShutdownAfter: 0,
-			MockDelay:     300 * time.Millisecond,
-			MockError:     fakesocket.ErrClosedConnection,
-			ExpectError:   fakesocket.ErrClosedConnection,
+			Scenario:        "Server failed throughout runtime and successfully committed to lambda",
+			EventStatus:     http.StatusOK,
+			TelemetryStatus: http.StatusOK,
+			ShutdownAfter:   0,
+			MockDelay:       300 * time.Millisecond,
+			MockError:       fakesocket.ErrClosedConnection,
+			ExpectError:     fakesocket.ErrClosedConnection,
+		},
+		{
+			Scenario:        "Server failed to subscribe to telemetry",
+			EventStatus:     http.StatusOK,
+			TelemetryStatus: http.StatusInternalServerError,
+			ShutdownAfter:   0,
+			MockDelay:       300 * time.Millisecond,
+			MockError:       nil,
+			ExpectError:     ErrFailedTelemetrySubscription,
 		},
 	}
 
@@ -281,18 +311,13 @@ func TestManagerDo(t *testing.T) {
 		tc := tc
 		t.Run(tc.Scenario, func(t *testing.T) {
 			t.Parallel()
-			clck := clock.NewMock(time.Unix(1, 0))
-
-			ctx, cancel := context.WithCancel(
-				clock.Context(context.Background(), clck),
-			)
-			t.Cleanup(cancel)
 
 			r := mux.NewRouter()
 			r.Handle(api.RegisterEndpoint.String(), InitHandler(t, http.StatusOK))
 			r.Handle(api.InitErrorEndpoint.String(), InitErrorHandler(t, http.StatusAccepted))
 			r.Handle(api.ExitErrorEndpoint.String(), ExitErrorHandler(t, http.StatusAccepted))
-			r.Handle(api.EventEndpoint.String(), EventNextHandler(t, ctx, tc.EventStatus, tc.ShutdownAfter))
+			r.Handle(telemetry.SubscribeEndpoint.String(), TelemetryHandler(t, tc.TelemetryStatus))
+			r.Handle(api.EventEndpoint.String(), EventNextHandler(t, tc.EventStatus, tc.ShutdownAfter))
 			r.PathPrefix("/").HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
 				assert.Fail(t, "Trying to access wrong path", r.RequestURI)
 			})
@@ -302,19 +327,22 @@ func TestManagerDo(t *testing.T) {
 
 			u, err := url.Parse(s.URL)
 			require.NoError(t, err, "Must be able to parse URL from httptest server")
+			log := logrus.New()
 
 			m := &manager{
-				log:    logrus.New(),
-				client: s.Client(),
-				domain: u.Hostname() + ":" + u.Port(),
-				name:   t.Name(),
+				log:             log,
+				client:          s.Client(),
+				domain:          u.Hostname() + ":" + u.Port(),
+				name:            t.Name(),
+				registeredID:    t.Name(),
+				telemetryServer: telemetry.NewServer(availableAddr(), log, telemetry.NoopHook()),
 			}
 
 			start := make(chan struct{}, 1)
 			done := make(chan struct{}, 1)
 
 			server := &mocked{delay: tc.MockDelay, erred: tc.MockError, start: start, done: done}
-
+			ctx, cancel := context.WithCancel(context.Background())
 			go func() {
 				<-start
 				close(start)
@@ -322,16 +350,64 @@ func TestManagerDo(t *testing.T) {
 					select {
 					case <-done:
 						close(done)
-						fixtures.NextStep(ctx, clck)
 						cancel()
 						return
-					default:
-						fixtures.NextStep(ctx, clck)
 					}
 				}
 			}()
 			assert.ErrorIs(t, m.Run(ctx, server), tc.ExpectError)
-			assert.Zero(t, clck.Len(), "Must have stopped all timers")
+		})
+	}
+}
+
+func TestManagerTelemetrySubscription(t *testing.T) {
+	t.Parallel()
+
+	testcases := []struct {
+		name          string
+		serverStatus  int
+		expectedError error
+	}{
+		{
+			name:          "Successful subscription",
+			serverStatus:  http.StatusOK,
+			expectedError: nil,
+		},
+		{
+			name:          "Bad subscription request",
+			serverStatus:  http.StatusBadRequest,
+			expectedError: ErrFailedTelemetrySubscription,
+		},
+		{
+			name:          "Container error",
+			serverStatus:  http.StatusInternalServerError,
+			expectedError: ErrFailedTelemetrySubscription,
+		},
+	}
+
+	for _, tc := range testcases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			r := mux.NewRouter()
+			r.Handle(telemetry.SubscribeEndpoint.String(), TelemetryHandler(t, tc.serverStatus))
+			s := httptest.NewServer(r)
+			t.Cleanup(s.Close)
+
+			u, err := url.Parse(s.URL)
+			require.NoError(t, err, "Must be able to parse URL from httptest server")
+			log := logrus.New()
+
+			m := &manager{
+				log:             log,
+				client:          s.Client(),
+				domain:          u.Hostname() + ":" + u.Port(),
+				name:            t.Name(),
+				registeredID:    t.Name(),
+				telemetryServer: telemetry.NewServer(availableAddr(), log, telemetry.NoopHook()),
+			}
+
+			assert.ErrorIs(t, tc.expectedError, m.subscribeToTelemetry(context.Background()))
 		})
 	}
 }
diff --git a/internal/awslambda/extension/telemetry/server_test.go b/internal/awslambda/extension/telemetry/server_test.go
new file mode 100644
index 0000000..575f897
--- /dev/null
+++ b/internal/awslambda/extension/telemetry/server_test.go
@@ -0,0 +1,91 @@
+package telemetry
+
+import (
+	"bytes"
+	"context"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	jsoniter "github.com/json-iterator/go"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewServer(t *testing.T) {
+	t.Parallel()
+
+	assert.NotNil(t, NewServer("test:80", logrus.New(), NoopHook()))
+}
+
+func TestServerStart(t *testing.T) {
+	t.Parallel()
+
+	l, _ := net.Listen("tcp", ":0")
+	l.Close()
+
+	s := NewServer(l.Addr().String(), logrus.New(), NoopHook())
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-time.After(1 * time.Second)
+		cancel()
+	}()
+	err := s.Start(ctx)
+	assert.NoError(t, err)
+}
+
+func TestRuntimeDoneHook(t *testing.T) {
+	t.Parallel()
+
+	tcs := []struct {
+		name       string
+		payload    []Event
+		hookCalled bool
+	}{
+		{
+			name:       "should invoke hook on runtimeDoneEvent",
+			payload:    []Event{{RuntimeDone}},
+			hookCalled: true,
+		},
+		{
+			name:       "should not invoke hook",
+			payload:    []Event{{"platform.init"}, {"platform.start"}},
+			hookCalled: false,
+		},
+	}
+
+	for _, tc := range tcs {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			callbackInvoked := false
+			f := func() { callbackInvoked = true }
+
+			p, err := jsoniter.Marshal(tc.payload)
+			require.NoError(t, err)
+
+			l, _ := net.Listen("tcp", ":0")
+			l.Close()
+
+			s := NewServer(l.Addr().String(), logrus.New(), f)
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			go s.Start(ctx)
+
+			res, err := http.Post(s.Endpoint(), "application/json", bytes.NewReader(p))
+			defer res.Body.Close()
+
+			assert.Equal(t, http.StatusOK, res.StatusCode)
+			assert.Equal(t, tc.hookCalled, callbackInvoked)
+		})
+	}
+}
+
+func TestServerReturnsCorrectEndpoint(t *testing.T) {
+	t.Parallel()
+	s := NewServer(LambdaRuntimeAvailableAddr, logrus.New(), NoopHook())
+	assert.Equal(t, "http://sandbox:8083/telemetry", s.Endpoint())
+}
diff --git a/pkg/backends/otlp/client_test.go b/pkg/backends/otlp/client_test.go
deleted file mode 100644
index 1c15a33..0000000
--- a/pkg/backends/otlp/client_test.go
+++ /dev/null
@@ -1,1 +0,0 @@
-package otlp
diff --git a/pkg/backends/otlp/internal/data/gauge_test.go b/pkg/backends/otlp/internal/data/gauge_test.go
deleted file mode 100644
index 3e52a72..0000000
--- a/pkg/backends/otlp/internal/data/gauge_test.go
+++ /dev/null
@@ -1,17 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestGauge(t *testing.T) {
-	t.Parallel()
-
-	g := NewGauge()
-	assert.Len(t, g.raw.DataPoints, 0, "Must have no datapoints defined")
-
-	g = NewGauge(NewNumberDataPoint(100), NewNumberDataPoint(100))
-	assert.Len(t, g.raw.DataPoints, 2, "Must have two datapoints defined")
-}
diff --git a/pkg/backends/otlp/internal/data/histograms_test.go b/pkg/backends/otlp/internal/data/histograms_test.go
deleted file mode 100644
index 3ffb471..0000000
--- a/pkg/backends/otlp/internal/data/histograms_test.go
+++ /dev/null
@@ -1,111 +0,0 @@
-package data
-
-import (
-	"math"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	v1common "go.opentelemetry.io/proto/otlp/common/v1"
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-func TestHistogram(t *testing.T) {
-	t.Parallel()
-
-	empty := NewHistogram()
-	assert.Len(t, empty.raw.DataPoints, 0, "Must have no datapoints defined")
-	assert.Equal(t,
-		v1metrics.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA,
-		empty.raw.AggregationTemporality,
-	)
-
-	h := NewHistogram(NewHistogramDataPoint(10), NewHistogramDataPoint(10))
-	assert.Len(t, h.raw.DataPoints, 2, "Must have two datapoints defined")
-}
-
-func TestHistogramDataPoint(t *testing.T) {
-	t.Parallel()
-
-	for _, tc := range []struct {
-		name string
-
-		opts   []func(HistogramDataPoint)
-		expect HistogramDataPoint
-	}{
-		{
-			name: "Empty",
-			expect: HistogramDataPoint{
-				raw: &v1metrics.HistogramDataPoint{
-					TimeUnixNano: 10,
-				},
-			},
-		},
-		{
-			name: "assigns metadata",
-			opts: []func(HistogramDataPoint){
-				WithHistogramDataPointAttributes(
-					NewMap(WithDelimitedStrings(":", []string{"service.name:my-awesome-service"})),
-				),
-			},
-			expect: HistogramDataPoint{
-				raw: &v1metrics.HistogramDataPoint{
-					TimeUnixNano: 10,
-					Attributes: []*v1common.KeyValue{
-						{
-							Key: "service.name",
-							Value: &v1common.AnyValue{
-								Value: &v1common.AnyValue_StringValue{
-									StringValue: "my-awesome-service",
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-		{
-			name: "populates values",
-			opts: []func(HistogramDataPoint){
-				WithHistogramDataPointStatistics(
-					[]float64{0.0, 1.2, 3.8},
-				),
-				WithHistogramDataPointBucketValues(map[float64]uint64{
-					1.0:         1,
-					1.4:         2,
-					3.0:         2,
-					4.0:         3,
-					math.Inf(1): 3,
-				}),
-			},
-			expect: func() HistogramDataPoint {
-				min, max, sum := 0.0, 3.8, 5.0
-				return HistogramDataPoint{
-					raw: &v1metrics.HistogramDataPoint{
-						Sum:          &sum,
-						Min:          &min,
-						Max:          &max,
-						Count:        3,
-						TimeUnixNano: 10,
-						BucketCounts: []uint64{
-							1, 2, 2, 3, 3,
-						},
-						ExplicitBounds: []float64{
-							1.0,
-							1.4,
-							3.0,
-							4.0,
-						},
-					},
-				}
-			}(),
-		},
-	} {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			dp := NewHistogramDataPoint(10, tc.opts...)
-			assert.Equal(t, tc.expect, dp, "Must match the expected value")
-		})
-	}
-}
diff --git a/pkg/backends/otlp/internal/data/instrumentation_scope_test.go b/pkg/backends/otlp/internal/data/instrumentation_scope_test.go
deleted file mode 100644
index 67f84ad..0000000
--- a/pkg/backends/otlp/internal/data/instrumentation_scope_test.go
+++ /dev/null
@@ -1,50 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	v1common "go.opentelemetry.io/proto/otlp/common/v1"
-)
-
-func TestInstrumentationScope(t *testing.T) {
-	t.Parallel()
-
-	is := NewInstrumentationScope("gostatsd", "test")
-	assert.Equal(
-		t,
-		is,
-		InstrumentationScope{
-			raw: &v1common.InstrumentationScope{
-				Name:    "gostatsd",
-				Version: "test",
-			},
-		},
-	)
-
-	is = NewInstrumentationScope("gostatsd", "test",
-		WithInstrumentationScopeAttributes(
-			NewMap(WithDelimitedStrings(":", []string{"service.name:my-awesome-service"})),
-		),
-	)
-	assert.Equal(
-		t,
-		is,
-		InstrumentationScope{
-			raw: &v1common.InstrumentationScope{
-				Name:    "gostatsd",
-				Version: "test",
-				Attributes: []*v1common.KeyValue{
-					{
-						Key: "service.name",
-						Value: &v1common.AnyValue{
-							Value: &v1common.AnyValue_StringValue{
-								StringValue: "my-awesome-service",
-							},
-						},
-					},
-				},
-			},
-		},
-	)
-}
diff --git a/pkg/backends/otlp/internal/data/map_test.go b/pkg/backends/otlp/internal/data/map_test.go
deleted file mode 100644
index 5b9ad9e..0000000
--- a/pkg/backends/otlp/internal/data/map_test.go
+++ /dev/null
@@ -1,91 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestNewMap(t *testing.T) {
-	t.Parallel()
-
-	for _, tc := range []struct {
-		name string
-		opts []func(Map)
-		m    Map
-	}{
-		{
-			name: "Blank map",
-			m:    NewMap(),
-		},
-		{
-			name: "String delim values",
-			opts: []func(Map){
-				WithDelimitedStrings(":",
-					[]string{
-						"service.name:my-awesome-service",
-						"service.version:1.0.0",
-						"service.namespace:internal/important:very",
-					},
-				),
-			},
-			m: NewMap(func(m Map) {
-				m.Insert("service.name", "my-awesome-service")
-				m.Insert("service.version", "1.0.0")
-				m.Insert("service.namespace", "internal/important:very")
-			}),
-		},
-		{
-			name: "Duplicate entries",
-			opts: []func(Map){
-				WithStatsdDelimitedTags([]string{
-					"service.name:my-awesome-service",
-					"service.name:very-awesome",
-				}),
-			},
-			m: NewMap(func(m Map) {
-				m.Insert("service.name", "my-awesome-service")
-				m.Insert("service.name", "very-awesome")
-			}),
-		},
-		{
-			name: "no delim",
-			opts: []func(Map){
-				WithDelimitedStrings("$", []string{"service.name:my-awesome-service"}),
-			},
-			m: NewMap(func(m Map) {
-				m.Insert("service.name:my-awesome-service", "")
-			}),
-		},
-	} {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			m := NewMap(tc.opts...)
-
-			if !assert.True(t, tc.m.Equal(m), "Must match the expected values") {
-				assert.Equal(t, tc.m, m)
-			}
-		})
-	}
-}
-
-func TestMapMerge(t *testing.T) {
-	t.Parallel()
-
-	empty := NewMap()
-	empty.Merge(empty)
-
-	assert.Len(t, *empty.raw, 0, "Must not have any values set")
-
-	m := NewMap(WithDelimitedStrings(":", []string{"service.name:my-awesome-service"}))
-	assert.Len(t, *m.raw, 1, "Must have one value defined")
-
-	empty.Merge(m)
-	assert.Len(t, *empty.raw, 1, "Must have values from contained")
-
-	empty.Merge(m)
-	assert.Len(t, *empty.raw, 1, "Must have values from contained")
-
-}
diff --git a/pkg/backends/otlp/internal/data/metric_request_test.go b/pkg/backends/otlp/internal/data/metric_request_test.go
deleted file mode 100644
index 482cc2a..0000000
--- a/pkg/backends/otlp/internal/data/metric_request_test.go
+++ /dev/null
@@ -1,25 +0,0 @@
-package data
-
-import (
-	"context"
-	"io"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestMetricsRequest(t *testing.T) {
-	t.Parallel()
-
-	req, err := NewMetricsRequest(
-		context.Background(),
-		"not-a-valid-url",
-		NewResourceMetrics(NewResource()),
-	)
-	assert.NoError(t, err, "Must not error creating request")
-	assert.NotNil(t, req, "Must have a valid request")
-
-	body, err := io.ReadAll(req.Body)
-	assert.NoError(t, err, "Must not error reading contents")
-	assert.NotEmpty(t, body, "Must have content set")
-}
diff --git a/pkg/backends/otlp/internal/data/metric_response_test.go b/pkg/backends/otlp/internal/data/metric_response_test.go
deleted file mode 100644
index f251786..0000000
--- a/pkg/backends/otlp/internal/data/metric_response_test.go
+++ /dev/null
@@ -1,63 +0,0 @@
-package data
-
-import (
-	"bytes"
-	"io"
-	"net/http"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	v1export "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
-	"google.golang.org/protobuf/proto"
-)
-
-func TestProcessMetricsResponse(t *testing.T) {
-	t.Parallel()
-
-	for _, tc := range []struct {
-		name    string
-		body    io.Reader
-		dropped int64
-		errVal  string
-	}{
-		{
-			name:    "Valid response",
-			body:    bytes.NewBuffer(nil),
-			dropped: 0,
-			errVal:  "",
-		},
-		{
-			name: "Dropped data",
-			body: func() io.Reader {
-				buf, err := proto.Marshal(&v1export.ExportMetricsServiceResponse{
-					PartialSuccess: &v1export.ExportMetricsPartialSuccess{
-						RejectedDataPoints: 12,
-						ErrorMessage:       "missing service name",
-					},
-				})
-				require.NoError(t, err, "Must not error when constructing test values")
-				return bytes.NewBuffer(buf)
-			}(),
-			dropped: 12,
-			errVal:  "dataloss: dropped 12 metrics; failed to send metrics: missing service name",
-		},
-	} {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			dropped, err := ProcessMetricResponse(
-				&http.Response{
-					Body: io.NopCloser(tc.body),
-				},
-			)
-			assert.Equal(t, tc.dropped, dropped)
-			if tc.errVal != "" {
-				assert.EqualError(t, err, tc.errVal, "Must match the expected value")
-			} else {
-				assert.NoError(t, err, "Must not error")
-			}
-		})
-	}
-}
diff --git a/pkg/backends/otlp/internal/data/metric_test.go b/pkg/backends/otlp/internal/data/metric_test.go
deleted file mode 100644
index df89548..0000000
--- a/pkg/backends/otlp/internal/data/metric_test.go
+++ /dev/null
@@ -1,33 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-func TestNewMetric(t *testing.T) {
-	t.Parallel()
-
-	g, s, h := NewGauge(), NewSum(), NewHistogram()
-
-	m := NewMetric("my-awesome-metric")
-	assert.Equal(t, "my-awesome-metric", m.raw.Name, "Must match the expected name")
-	assert.Nil(t, m.raw.Data, "No metric value set")
-
-	m.SetGauge(g)
-	assert.Equal(t, m.raw.Data, &v1metrics.Metric_Gauge{
-		Gauge: g.raw,
-	})
-
-	m.SetSum(s)
-	assert.Equal(t, m.raw.Data, &v1metrics.Metric_Sum{
-		Sum: s.raw,
-	})
-
-	m.SetHistogram(h)
-	assert.Equal(t, m.raw.Data, &v1metrics.Metric_Histogram{
-		Histogram: h.raw,
-	})
-}
diff --git a/pkg/backends/otlp/internal/data/number_datapoint_test.go b/pkg/backends/otlp/internal/data/number_datapoint_test.go
deleted file mode 100644
index 3e7355a..0000000
--- a/pkg/backends/otlp/internal/data/number_datapoint_test.go
+++ /dev/null
@@ -1,86 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	v1common "go.opentelemetry.io/proto/otlp/common/v1"
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-func TestNumberDataPoint(t *testing.T) {
-	t.Parallel()
-
-	for _, tc := range []struct {
-		name   string
-		opts   []func(NumberDataPoint)
-		expect NumberDataPoint
-	}{
-		{
-			name: "empty",
-			expect: NumberDataPoint{
-				raw: &v1metrics.NumberDataPoint{
-					TimeUnixNano: 100,
-				},
-			},
-		},
-		{
-			name: "int with values",
-			opts: []func(NumberDataPoint){
-				WithNumberDataPointDelimtedTags([]string{"service.name:my-awesome-service"}),
-				WithNumberDatapointIntValue(1),
-			},
-			expect: NumberDataPoint{
-				raw: &v1metrics.NumberDataPoint{
-					TimeUnixNano: 100,
-					Attributes: []*v1common.KeyValue{
-						{
-							Key: "service.name",
-							Value: &v1common.AnyValue{
-								Value: &v1common.AnyValue_StringValue{
-									StringValue: "my-awesome-service",
-								},
-							},
-						},
-					},
-					Value: &v1metrics.NumberDataPoint_AsInt{
-						AsInt: 1,
-					},
-				},
-			},
-		},
-		{
-			name: "double with values",
-			opts: []func(NumberDataPoint){
-				WithNumberDataPointDelimtedTags([]string{"service.name:my-awesome-service"}),
-				WithNumberDataPointDoubleValue(1),
-			},
-			expect: NumberDataPoint{
-				raw: &v1metrics.NumberDataPoint{
-					TimeUnixNano: 100,
-					Attributes: []*v1common.KeyValue{
-						{
-							Key: "service.name",
-							Value: &v1common.AnyValue{
-								Value: &v1common.AnyValue_StringValue{
-									StringValue: "my-awesome-service",
-								},
-							},
-						},
-					},
-					Value: &v1metrics.NumberDataPoint_AsDouble{
-						AsDouble: 1,
-					},
-				},
-			},
-		},
-	} {
-		tc := tc
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-
-			dp := NewNumberDataPoint(100, tc.opts...)
-			assert.Equal(t, dp, tc.expect, "Must match the expected value")
-		})
-	}
-}
diff --git a/pkg/backends/otlp/internal/data/resource_metrics_test.go b/pkg/backends/otlp/internal/data/resource_metrics_test.go
deleted file mode 100644
index 56438ef..0000000
--- a/pkg/backends/otlp/internal/data/resource_metrics_test.go
+++ /dev/null
@@ -1,31 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	v1resource "go.opentelemetry.io/proto/otlp/resource/v1"
-)
-
-func TestNewResourceMetrics(t *testing.T) {
-	t.Parallel()
-
-	empty := NewResourceMetrics(NewResource())
-	assert.Equal(
-		t,
-		&v1resource.Resource{},
-		empty.raw.Resource,
-	)
-	assert.Len(t, empty.raw.ScopeMetrics, 0, "Must have no values set")
-
-	rm := NewResourceMetrics(
-		NewResource(),
-		NewScopeMetrics(
-			NewInstrumentationScope("gostatsd/aggregation", "test"),
-		),
-		NewScopeMetrics(
-			NewInstrumentationScope("gostatsd/aggregation", "test"),
-		),
-	)
-	assert.Len(t, rm.raw.ScopeMetrics, 2, "Must have two scope metrics defined")
-}
diff --git a/pkg/backends/otlp/internal/data/resource_test.go b/pkg/backends/otlp/internal/data/resource_test.go
deleted file mode 100644
index 82be791..0000000
--- a/pkg/backends/otlp/internal/data/resource_test.go
+++ /dev/null
@@ -1,21 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestNewResource(t *testing.T) {
-	t.Parallel()
-
-	empty := NewResource()
-	assert.Len(t, empty.raw.Attributes, 0, "Must have no values defined")
-
-	r := NewResource(
-		WithResourceAttributes(
-			WithDelimitedStrings(":", []string{"service.name:my-awesome-service"}),
-		),
-	)
-	assert.Len(t, r.raw.Attributes, 1, "must have one value defined")
-}
diff --git a/pkg/backends/otlp/internal/data/scope_metrics_test.go b/pkg/backends/otlp/internal/data/scope_metrics_test.go
deleted file mode 100644
index c8618b3..0000000
--- a/pkg/backends/otlp/internal/data/scope_metrics_test.go
+++ /dev/null
@@ -1,33 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-func TestNewScopeMetrics(t *testing.T) {
-	t.Parallel()
-
-	empty := NewScopeMetrics(
-		NewInstrumentationScope("gostatsd/aggregation", "test"),
-	)
-	assert.Equal(
-		t,
-		ScopeMetrics{
-			raw: &v1metrics.ScopeMetrics{
-				Scope:   NewInstrumentationScope("gostatsd/aggregation", "test").raw,
-				Metrics: make([]*v1metrics.Metric, 0),
-			},
-		},
-		empty,
-	)
-
-	sm := NewScopeMetrics(
-		NewInstrumentationScope("gostatsd/aggregation", "test"),
-		NewMetric("my-awesome-metric"),
-	)
-	assert.Len(t, sm.raw.Metrics, 1, "Must have one metric set")
-	assert.Equal(t, sm.raw.Metrics[0].Name, "my-awesome-metric")
-}
diff --git a/pkg/backends/otlp/internal/data/sum_test.go b/pkg/backends/otlp/internal/data/sum_test.go
deleted file mode 100644
index 39aa7e9..0000000
--- a/pkg/backends/otlp/internal/data/sum_test.go
+++ /dev/null
@@ -1,28 +0,0 @@
-package data
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-	v1metrics "go.opentelemetry.io/proto/otlp/metrics/v1"
-)
-
-func TestNewSum(t *testing.T) {
-	t.Parallel()
-
-	empty := NewSum()
-	assert.Equal(
-		t,
-		Sum{
-			raw: &v1metrics.Sum{
-				IsMonotonic:            false,
-				AggregationTemporality: v1metrics.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA,
-				DataPoints:             make([]*v1metrics.NumberDataPoint, 0),
-			},
-		},
-		empty,
-	)
-
-	s := NewSum(NewNumberDataPoint(100), NewNumberDataPoint(100))
-	assert.Len(t, s.raw.DataPoints, 2, "Must have two datapoints set")
-}