Add a wrapper for vpp connections with extended timeout

extendtimeout/connection.go

@@ -0,0 +1,74 @@
+// Copyright (c) 2024 Cisco and/or its affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// 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 extendtimeout - provides a wrapper for vpp connection that uses extended timeout for all vpp operations
+package extendtimeout
+
+import (
+	"context"
+	"time"
+
+	"github.com/edwarnicke/log"
+	"go.fd.io/govpp/api"
+)
+
+type extendedConnection struct {
+	api.Connection
+	contextTimeout time.Duration
+}
+
+type extendedContext struct {
+	context.Context
+	valuesContext context.Context
+}
+
+func (ec *extendedContext) Value(key interface{}) interface{} {
+	return ec.valuesContext.Value(key)
+}
+
+// NewConnection - creates a wrapper for vpp connection that uses extended context timeout for all operations
+func NewConnection(vppConn api.Connection, contextTimeout time.Duration) api.Connection {
+	return &extendedConnection{
+		Connection:     vppConn,
+		contextTimeout: contextTimeout,
+	}
+}
+
+func (c *extendedConnection) Invoke(ctx context.Context, req, reply api.Message) error {
+	ctx, cancel := c.withExtendedTimeoutCtx(ctx)
+	err := c.Connection.Invoke(ctx, req, reply)
+	cancel()
+	return err
+}
+
+func (c *extendedConnection) withExtendedTimeoutCtx(ctx context.Context) (extendedCtx context.Context, cancel func()) {
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		return ctx, func() {}
+	}
+
+	minDeadline := time.Now().Add(c.contextTimeout)
+	if minDeadline.Before(deadline) {
+		return ctx, func() {}
+	}
+	log.Entry(ctx).Warnf("Context deadline has been extended by extendtimeout from %v to %v", deadline, minDeadline)
+	deadline = minDeadline
+	postponedCtx, cancel := context.WithDeadline(context.Background(), deadline)
+	return &extendedContext{
+		Context:       postponedCtx,
+		valuesContext: ctx,
+	}, cancel
+}

extendtimeout/connection_test.go

@@ -0,0 +1,120 @@
+// Copyright (c) 2024 Cisco and/or its affiliates.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// 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 extendtimeout_test
+
+import (
+	"context"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.fd.io/govpp/api"
+	"go.uber.org/goleak"
+
+	"github.com/networkservicemesh/vpphelper/extendtimeout"
+)
+
+type testConn struct {
+	api.Connection
+	invokeBody func(ctx context.Context)
+}
+
+func (c *testConn) Invoke(ctx context.Context, req, reply api.Message) error {
+	c.invokeBody(ctx)
+	return nil
+}
+
+type key struct{}
+
+const value = "value"
+
+func TestSmallTimeout(t *testing.T) {
+	testConn := &testConn{invokeBody: func(ctx context.Context) {
+		deadline, ok := ctx.Deadline()
+		require.True(t, ok)
+		timeout := time.Until(deadline)
+		require.Greater(t, timeout, time.Second)
+		require.Equal(t, ctx.Value(&key{}), value)
+	}}
+
+	smallCtx, cancel := context.WithTimeout(context.Background(), time.Millisecond*10)
+	smallCtx = context.WithValue(smallCtx, &key{}, value)
+	defer cancel()
+
+	err := extendtimeout.NewConnection(testConn, 2*time.Second).Invoke(smallCtx, nil, nil)
+	require.NoError(t, err)
+}
+
+func TestBigTimeout(t *testing.T) {
+	testConn := &testConn{invokeBody: func(ctx context.Context) {
+		deadline, ok := ctx.Deadline()
+		require.True(t, ok)
+		timeout := time.Until(deadline)
+		require.Greater(t, timeout, 7*time.Second)
+		require.Equal(t, ctx.Value(&key{}), value)
+	}}
+
+	bigCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	bigCtx = context.WithValue(bigCtx, &key{}, value)
+	defer cancel()
+
+	err := extendtimeout.NewConnection(testConn, 2*time.Second).Invoke(bigCtx, nil, nil)
+	require.NoError(t, err)
+}
+
+func TestNoTimeout(t *testing.T) {
+	testConn := &testConn{invokeBody: func(ctx context.Context) {
+		_, ok := ctx.Deadline()
+		require.False(t, ok)
+	}}
+
+	emptyCtx := context.Background()
+	err := extendtimeout.NewConnection(testConn, 2*time.Second).Invoke(emptyCtx, nil, nil)
+	require.NoError(t, err)
+}
+
+func TestCanceledContext(t *testing.T) {
+	t.Cleanup(func() {
+		goleak.VerifyNone(t)
+	})
+
+	counter := new(atomic.Int32)
+	ch := make(chan struct{}, 1)
+	defer close(ch)
+	testConn := &testConn{invokeBody: func(ctx context.Context) {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ch:
+			counter.Add(1)
+		}
+	}}
+
+	cancelCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	go func() {
+		err := extendtimeout.NewConnection(testConn, 20*time.Second).Invoke(cancelCtx, nil, nil)
+		require.NoError(t, err)
+	}()
+
+	cancel()
+	ch <- struct{}{}
+
+	require.Eventually(t, func() bool {
+		return counter.Load() == 1
+	}, time.Second, 100*time.Millisecond)
+}

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.8.4
 	go.fd.io/govpp v0.10.0-alpha.0.20240110141843-761adec77524
+	go.uber.org/goleak v1.3.0
 	gopkg.in/fsnotify.v1 v1.4.7
 )
 
@@ -19,6 +20,7 @@ require (
 	github.com/lunixbochs/struc v0.0.0-20200521075829-a4cb8d33dbbe // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
-	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/net v0.21.0 // indirect
+	golang.org/x/sys v0.18.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -35,12 +35,14 @@ github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcU
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 go.fd.io/govpp v0.10.0-alpha.0.20240110141843-761adec77524 h1:Dja0i7Ln/aUAc10VN4Pp3SibYTfNKS1CPkM5TSGL5jk=
 go.fd.io/govpp v0.10.0-alpha.0.20240110141843-761adec77524/go.mod h1:9QoqjEbvfuuXNfjHS0A7YS+7QQVVaQ9cMioOWpSM4rY=
-golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
-golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=
+golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
-golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
+golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=