Commit 610c3424 authored by Mudassar Khan's avatar Mudassar Khan
Browse files

add retry logic to avoid needing requests all at once and also to deal timeout errors

parent dcb02f5c
Loading
Loading
Loading
Loading
+6 −1
Changes for go-apps/meep-ipe/entrypoint.sh: 6 added lines, 1 removed line.
Original line number Diff line number Diff line
#!/bin/bash
set -e

sleep 10 # Wait for the MEC platform to stabilize and acme in cse to be ready
# Wait for the MEC platform to stabilize and acme in cse to be ready. A random
# 0-9s offset is added so MEC IPE instances started together in the cluster do
# not all send their first requests at the same moment.
STARTUP_DELAY=$((2 + RANDOM % 10))
echo "Waiting ${STARTUP_DELAY}s before starting meep-ipe"
sleep ${STARTUP_DELAY}

# BASE_PATH — normally derived from sandbox/location (used in the k8s/Helm
# deployment). For standalone/local docker run testing, set
+19 −16
Changes for go-apps/meep-ipe/utils/mec_req.go: 19 added lines, 16 removed lines.
Original line number Diff line number Diff line
@@ -6,6 +6,7 @@ import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"time"
@@ -16,6 +17,8 @@ import (
// This function creates and sends an HTTP request to a MEC platform API endpoint.
// It handles JSON payload marshaling for POST/PUT requests and sets appropriate
// headers for MEC API communication. TLS verification is disabled for development/testing.
// Timeouts, connection errors and 408/429/5xx responses are retried with
// exponential backoff and jitter (see doWithRetry).
//
// @param method HTTP method (GET, POST, PUT, DELETE, etc.)
// @param url The complete URL endpoint for the MEC API request
@@ -30,32 +33,32 @@ func SendMECRequest(method, url string, payload interface{}) (*http.Response, er
		},
	}

	var req *http.Request
	// Marshal the payload once; the request is rebuilt on each retry attempt
	var reqBody []byte
	if !(method == "GET" && payload == nil) {
		var err error

	// Handle GET requests with no payload
	if method == "GET" && payload == nil {
		req, err = http.NewRequestWithContext(context.Background(), method, url, nil)
		if err != nil {
			return nil, fmt.Errorf("failed to create request: %v", err)
		}
	} else {
		// Handle requests with payload
		reqBody, err := json.Marshal(payload)
		reqBody, err = json.Marshal(payload)
		if err != nil {
			return nil, fmt.Errorf("failed to marshal payload: %v", err)
		}
		log.Println("Request body:", string(reqBody))
	}

		req, err = http.NewRequestWithContext(context.Background(), method, url, bytes.NewBuffer(reqBody))
	// Transient failures are retried with jittered backoff (see retry.go)
	return doWithRetry(client, func() (*http.Request, error) {
		var body io.Reader
		if reqBody != nil {
			body = bytes.NewReader(reqBody)
		}
		req, err := http.NewRequestWithContext(context.Background(), method, url, body)
		if err != nil {
			return nil, fmt.Errorf("failed to create request: %v", err)
			return nil, err
		}
		if reqBody != nil {
			req.Header.Set("Content-Type", "application/json")
		}

		req.Header.Set("Accept", "application/json")
		req.Header.Set("User-Agent", "Go-http-client/1.1")

	return client.Do(req)
		return req, nil
	})
}
+8 −4
Changes for go-apps/meep-ipe/utils/onem2m_req.go: 8 added lines, 4 removed lines.
Original line number Diff line number Diff line
@@ -25,6 +25,8 @@ func SetCSECredentials(user, password string) {
// This function creates and sends an HTTP request to a oneM2M CSE platform endpoint.
// It sets the required oneM2M headers including X-M2M-Origin, X-M2M-RI (request ID),
// X-M2M-RVI (release version indicator), and Content-Type with resource type (ty).
// Timeouts, connection errors and 408/429/5xx responses are retried with
// exponential backoff and jitter (see doWithRetry).
//
// Reference: oneM2M TS-0001 - Functional Architecture
//
@@ -63,16 +65,18 @@ func MakeOneM2MRequest(method, url, body string, ty int, originator string) (*ht
		headers["Content-Type"] = "application/json"
	}

	// Transient failures are retried with jittered backoff (see retry.go); the
	// request is rebuilt on each attempt so the body can be resent
	resp, err := doWithRetry(client, func() (*http.Request, error) {
		req, err := http.NewRequest(method, url, strings.NewReader(body))
		if err != nil {
		return nil, fmt.Errorf("failed to create request: %v", err)
			return nil, err
		}

		for key, value := range headers {
			req.Header.Set(key, value)
		}

	resp, err := client.Do(req)
		return req, nil
	})
	if err != nil {
		return nil, fmt.Errorf("failed to send request: %v", err)
	}
+97 −0
Changes for go-apps/meep-ipe/utils/retry.go: 97 added lines, 0 removed lines.
Original line number Diff line number Diff line
package utils

import (
	"fmt"
	"io"
	"log"
	"math/rand/v2"
	"net/http"
	"strconv"
	"time"
)

// Retry policy for outbound requests (MEC platform and oneM2M CSE).
//
// Many MEC IPE instances in a cluster typically start at the same moment and hit
// the same MEC platform / CSE. Retrying on a fixed schedule would keep them all
// firing in lockstep, so the delay between attempts uses exponential backoff with
// "full jitter": each retry sleeps a random duration in [0, min(retryMaxDelay,
// retryBaseDelay*2^attempt)]. math/rand/v2 is seeded per process, so every
// instance draws a different delay and the retries spread out over time.
const (
	retryMaxAttempts = 5                      // total attempts, including the first one
	retryBaseDelay   = 500 * time.Millisecond // backoff ceiling for the first retry
	retryMaxDelay    = 8 * time.Second        // upper bound of the backoff ceiling
	retryMaxElapsed  = 45 * time.Second       // no new attempt is started after this
)

// doWithRetry sends the request produced by newReq, retrying on transport errors
// (timeouts, connection refused/reset, DNS failures, ...) and on responses that
// signal a transient server condition (408, 429, 5xx). Other responses, including
// 4xx client errors, are returned to the caller as-is.
//
// newReq is called once per attempt so that the request body can be replayed.
func doWithRetry(client *http.Client, newReq func() (*http.Request, error)) (*http.Response, error) {
	start := time.Now()

	for attempt := 0; ; attempt++ {
		req, err := newReq()
		if err != nil {
			return nil, fmt.Errorf("failed to create request: %v", err)
		}

		resp, err := client.Do(req)
		retryable, reason := shouldRetry(resp, err)
		if !retryable {
			return resp, err
		}

		delay := backoffDelay(attempt, resp)
		if attempt+1 >= retryMaxAttempts || time.Since(start)+delay > retryMaxElapsed {
			log.Printf("%s %s: giving up after %d attempt(s): %s", req.Method, req.URL, attempt+1, reason)
			return resp, err
		}

		// Release the connection of the response we are discarding
		if resp != nil {
			_, _ = io.Copy(io.Discard, resp.Body)
			resp.Body.Close()
		}

		log.Printf("%s %s: attempt %d/%d failed (%s), retrying in %v",
			req.Method, req.URL, attempt+1, retryMaxAttempts, reason, delay.Round(time.Millisecond))
		time.Sleep(delay)
	}
}

// shouldRetry reports whether a request outcome is transient and worth retrying.
func shouldRetry(resp *http.Response, err error) (bool, string) {
	if err != nil {
		return true, err.Error()
	}
	switch {
	case resp.StatusCode == http.StatusRequestTimeout,
		resp.StatusCode == http.StatusTooManyRequests,
		resp.StatusCode >= 500:
		return true, resp.Status
	}
	return false, ""
}

// backoffDelay returns a full-jitter exponential backoff delay for the given
// attempt. If the server sent a Retry-After header (in seconds), it is used as a
// lower bound and jitter is added on top so instances still do not resync.
func backoffDelay(attempt int, resp *http.Response) time.Duration {
	ceiling := retryBaseDelay << attempt
	if ceiling <= 0 || ceiling > retryMaxDelay {
		ceiling = retryMaxDelay
	}
	delay := rand.N(ceiling + 1)

	if resp != nil {
		if secs, err := strconv.Atoi(resp.Header.Get("Retry-After")); err == nil && secs > 0 {
			delay += time.Duration(secs) * time.Second
		}
	}
	return delay
}