feat: implement aws timeouts

aws/route_table.go

@@ -36,33 +36,7 @@ func (cli *Client) RouteTableExists(ctx context.Context, routeTable string) (boo
 	return true, nil
 }
 
-func (cli *Client) UpdateRouteTable(
-	ctx context.Context,
-	routeTable string,
-	cidr string,
-	networkInterfaceID string,
-) error {
-	// check if the route is already set
-	route, err := cli.findRoute(ctx, routeTable, cidr)
-	if err != nil {
-		return err
-	}
-
-	if route != nil && aws.ToString(route.NetworkInterfaceId) == networkInterfaceID {
-		// route is already up to date
-		return nil
-	}
-
-	if route != nil {
-		// route exists but with different next hop
-		return cli.updateRoute(ctx, routeTable, cidr, networkInterfaceID)
-	}
-
-	// no route yet
-	return cli.createRoute(ctx, routeTable, cidr, networkInterfaceID)
-}
-
-func (cli *Client) findRoute(
+func (cli *Client) FindRoute(
 	ctx context.Context,
 	routeTable string,
 	cidr string,
@@ -100,7 +74,7 @@ func (cli *Client) findRoute(
 	return nil, nil
 }
 
-func (cli *Client) updateRoute(
+func (cli *Client) UpdateRoute(
 	ctx context.Context,
 	routeTable string,
 	cidr string,
@@ -130,7 +104,7 @@ func (cli *Client) updateRoute(
 	return err
 }
 
-func (cli *Client) createRoute(
+func (cli *Client) CreateRoute(
 	ctx context.Context,
 	routeTable string,
 	cidr string,

config/defaults.go

@@ -14,6 +14,7 @@ const (
 	DefaultThresholdDown = 5
 	DefaultThresholdUp   = 2
 
+	DefaultAWSTimeout     = 15 * time.Second
 	DefaultScriptsTimeout = 30 * time.Second
 
 	DefaultMetricsListenAddr = "0.0.0.0:8000"

config/reconcile_bridge_activate.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/flashbots/vpnham/aws"
 	"github.com/flashbots/vpnham/types"
+	"github.com/flashbots/vpnham/utils"
 )
 
 type ReconcileBridgeActivate struct {
@@ -22,6 +24,8 @@ type ReconcileBridgeActivateAWS struct {
 	NetworkInterfaceID string `yaml:"-"`
 	Region             string `yaml:"-"`
 
+	Timeout time.Duration `yaml:"timeout"`
+
 	RouteTables []string `yaml:"route_tables"`
 }
 
@@ -39,17 +43,33 @@ func (r *ReconcileBridgeActivate) PostLoad(ctx context.Context) error {
 	}
 
 	if r.AWS != nil {
-		reg, err := aws.Region(ctx)
-		if err != nil {
-			return err
+		if r.AWS.Timeout == 0 {
+			r.AWS.Timeout = DefaultAWSTimeout
 		}
-		r.AWS.Region = reg
 
-		eni, err := aws.NetworkInterfaceId(ctx, r.BridgeInterface)
-		if err != nil {
-			return err
+		{ // aws region
+			var reg string
+			err := utils.WithTimeout(ctx, r.AWS.Timeout, func(ctx context.Context) (err error) {
+				reg, err = aws.Region(ctx)
+				return err
+			})
+			if err != nil {
+				return err
+			}
+			r.AWS.Region = reg
+		}
+
+		{ // aws ec2 network interface id
+			var networkInterfaceID string
+			err := utils.WithTimeout(ctx, r.AWS.Timeout, func(ctx context.Context) (err error) {
+				networkInterfaceID, err = aws.NetworkInterfaceId(ctx, r.BridgeInterface)
+				return err
+			})
+			if err != nil {
+				return err
+			}
+			r.AWS.NetworkInterfaceID = networkInterfaceID
 		}
-		r.AWS.NetworkInterfaceID = eni
 	}
 
 	return nil

job/aws.go

@@ -0,0 +1,106 @@
+package job
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	awssdk "github.com/aws/aws-sdk-go-v2/aws"
+	awstypes "github.com/aws/aws-sdk-go-v2/service/ec2/types"
+	"github.com/flashbots/vpnham/aws"
+	"github.com/flashbots/vpnham/metrics"
+	"github.com/flashbots/vpnham/utils"
+	"go.opentelemetry.io/otel/attribute"
+	otelapi "go.opentelemetry.io/otel/metric"
+)
+
+type updateAWSRouteTables struct {
+	name    string
+	timeout time.Duration
+
+	cidr               string
+	networkInterfaceID string
+	routeTables        []string
+}
+
+func UpdateAWSRouteTables(
+	name string,
+	timeout time.Duration,
+	cidr string,
+	networkInterfaceID string,
+	routeTables []string,
+) Job {
+	return &updateAWSRouteTables{
+		name:               name,
+		timeout:            timeout,
+		cidr:               cidr,
+		networkInterfaceID: networkInterfaceID,
+		routeTables:        routeTables,
+	}
+}
+
+func (j *updateAWSRouteTables) Name() string {
+	return j.name
+}
+
+func (j *updateAWSRouteTables) Execute(ctx context.Context) error {
+	errs := []error{}
+	for _, rt := range j.routeTables {
+		err := j.updateRouteTable(ctx, rt, j.cidr, j.networkInterfaceID)
+		if err != nil {
+			metrics.Errors.Add(ctx, 1, otelapi.WithAttributes(
+				attribute.String(metrics.LabelErrorScope, "job_"+j.name),
+			))
+			errs = append(errs, err)
+		}
+	}
+
+	switch len(errs) {
+	default:
+		return errors.Join(errs...)
+	case 1:
+		return errs[0]
+	case 0:
+		return nil
+	}
+}
+
+func (j *updateAWSRouteTables) updateRouteTable(
+	ctx context.Context,
+	routeTable string,
+	cidr string,
+	networkInterfaceID string,
+) error {
+	cli, err := aws.NewClient(ctx)
+	if err != nil {
+		return err
+	}
+
+	var route *awstypes.Route
+
+	// check if the route is already set
+	err = utils.WithTimeout(ctx, j.timeout, func(ctx context.Context) error {
+		route, err = cli.FindRoute(ctx, routeTable, cidr)
+		return err
+	})
+	if err != nil {
+		return err
+	}
+
+	if route != nil && awssdk.ToString(route.NetworkInterfaceId) == networkInterfaceID {
+		// route is already up to date
+		return nil
+	}
+
+	if route != nil {
+		// route exists but with different next hop
+		return utils.WithTimeout(ctx, j.timeout, func(ctx context.Context) error {
+			return cli.UpdateRoute(ctx, routeTable, cidr, networkInterfaceID)
+		})
+	}
+
+	// no route yet
+	return utils.WithTimeout(ctx, j.timeout, func(ctx context.Context) error {
+		return cli.CreateRoute(ctx, routeTable, cidr, networkInterfaceID)
+	})
+}

job/job.go

@@ -0,0 +1,8 @@
+package job
+
+import "context"
+
+type Job interface {
+	Execute(context.Context) error
+	Name() string
+}

job/script.go

@@ -0,0 +1,105 @@
+package job
+
+import (
+	"context"
+	"errors"
+	"os"
+	"os/exec"
+	"strings"
+	"time"
+
+	"github.com/flashbots/vpnham/logutils"
+	"github.com/flashbots/vpnham/metrics"
+	"github.com/flashbots/vpnham/types"
+	"github.com/flashbots/vpnham/utils"
+	"go.opentelemetry.io/otel/attribute"
+	otelapi "go.opentelemetry.io/otel/metric"
+	"go.uber.org/zap"
+)
+
+type runScript struct {
+	name    string
+	timeout time.Duration
+
+	script types.Script
+}
+
+func RunScript(
+	name string,
+	timeout time.Duration,
+	script types.Script,
+) Job {
+	return &runScript{
+		name:    name,
+		timeout: timeout,
+		script:  script,
+	}
+}
+
+func (j *runScript) Name() string {
+	return j.name
+}
+
+func (j *runScript) Execute(ctx context.Context) error {
+	l := logutils.LoggerFromContext(ctx)
+
+	errs := []error{}
+	for step, _cmd := range j.script {
+		if len(_cmd) == 0 {
+			continue
+		}
+
+		strCmd := strings.Join(_cmd, " ")
+
+		l.Debug("Executing command",
+			zap.String("command", strCmd),
+		)
+
+		ctx, cancel := context.WithTimeout(ctx, j.timeout)
+		defer cancel()
+
+		cmd := exec.CommandContext(ctx, _cmd[0], _cmd[1:]...)
+
+		stdout := &strings.Builder{}
+		cmd.Stdout = stdout
+
+		stderr := &strings.Builder{}
+		cmd.Stderr = stderr
+
+		cmd.Env = os.Environ()
+
+		start := time.Now()
+		err := utils.WithTimeout(ctx, j.timeout, func(ctx context.Context) error {
+			return cmd.Run()
+		})
+		duration := time.Since(start)
+
+		if err != nil {
+			metrics.Errors.Add(ctx, 1, otelapi.WithAttributes(
+				attribute.String(metrics.LabelErrorScope, "job_"+j.name),
+			))
+			errs = append(errs, err)
+		}
+
+		l.Info("Executed command",
+			zap.String("script", j.name),
+			zap.Int("step", step),
+			zap.String("command", strCmd),
+			zap.Int64("duration_us", duration.Microseconds()),
+
+			zap.String("stderr", strings.TrimSpace(stderr.String())),
+			zap.String("stdout", strings.TrimSpace(stdout.String())),
+
+			zap.Error(err),
+		)
+	}
+
+	switch len(errs) {
+	default:
+		return errors.Join(errs...)
+	case 1:
+		return errs[0]
+	case 0:
+		return nil
+	}
+}

metrics/labels.go

@@ -11,7 +11,6 @@ const (
 )
 
 const (
-	ScopeAWS            = "aws"
 	ScopeHTTPMiddleware = "http_middleware"
 	ScopeInternalLogic  = "internal_logic"
 	ScopePartnerPolling = "partner_polling"