Stefan/evcc-io
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

OCPP Forwarder: reconnect upstream with backoff and replay BootNotification (#30765)

Browse source
This commit is contained in:
andig 2026-06-24 10:58:53 +02:00 • committed by GitHub
parent 0d9189d64c
commit 1b08a3461e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 322 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

202
charger/ocpp/forwarder.go
View file

@ -15,6 +15,7 @@ import (
"crypto/x509"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"net/http"
"slices"
@ -23,6 +24,7 @@ import (
"sync"
"time"
"github.com/cenkalti/backoff/v4"
"github.com/coder/websocket"
"github.com/evcc-io/evcc/util"
"github.com/lorenzodonini/ocpp-go/ocpp1.6/core"
@ -102,7 +104,7 @@ func ApplyForwarderRules(rules []ForwarderRule) {
pendingMu.Lock()
pendingMsgs[r.StationID] = nil
pendingMu.Unlock()
go dialUpstreamSidecar(r.StationID, r)
go runUpstreamSidecar(r.StationID, r)
} else {
go validateUpstream(r.StationID, r)
}
@ -111,19 +113,13 @@ func ApplyForwarderRules(rules []ForwarderRule) {
notifyUpdated()
}
// validateUpstream test-dials a rule's upstream and records/clears the error so
// the UI reflects unreachable hosts.
func validateUpstream(id string, rule ForwarderRule) {
upstreamBase := strings.TrimRight(rule.UpstreamURL, "/")
upstreamPath := rule.upstreamPath(id)
// dialUpstream opens a websocket connection to the rule's upstream server for the given charger.
func dialUpstream(id string, rule ForwarderRule) (*websocket.Conn, error) {
tlsConfig := &tls.Config{InsecureSkipVerify: rule.Insecure}
if rule.CaCert != "" {
caCertPool := x509.NewCertPool()
if ok := caCertPool.AppendCertsFromPEM([]byte(rule.CaCert)); !ok {
recordForwarderError(id, "invalid CA certificate")
notifyUpdated()
return
return nil, errors.New("invalid CA certificate")
}
tlsConfig.RootCAs = caCertPool
}
@ -134,12 +130,20 @@ func validateUpstream(id string, rule ForwarderRule) {
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
conn, _, err := websocket.Dial(ctx, upstreamBase+upstreamPath, &websocket.DialOptions{
defer cancel()
conn, _, err := websocket.Dial(ctx, strings.TrimRight(rule.UpstreamURL, "/")+rule.upstreamPath(id), &websocket.DialOptions{
Subprotocols: []string{"ocpp1.6"},
HTTPHeader: header,
HTTPClient: &http.Client{Transport: &http.Transport{TLSClientConfig: tlsConfig}},
})
cancel()
return conn, err
}
// validateUpstream test-dials a rule's upstream and records/clears the error so
// the UI reflects unreachable hosts.
func validateUpstream(id string, rule ForwarderRule) {
conn, err := dialUpstream(id, rule)
// a charger may have connected meanwhile; its sidecar is authoritative
sidecarsMu.Lock()
@ -190,10 +194,9 @@ func init() {
// sidecar holds an upstream connection for a single charger.
type sidecar struct {
chargerID string
upstreamURL string
rule ForwarderRule // rule used to dial; detects param changes
conn *websocket.Conn
chargerID string
rule ForwarderRule // rule used to dial; detects param changes
conn *websocket.Conn
// message IDs of upstream-initiated Calls; the charger's reply is routed back to upstream
pendingUpstreamCallsMu sync.Mutex
@ -223,8 +226,58 @@ var (
// on connect so BootNotification reaches upstream
pendingMu sync.Mutex
pendingMsgs = make(map[string][][]byte)
// last BootNotification per connected charger, replayed to upstream when a
// sidecar (re)connects mid-session so upstream sees a boot before transactions
bootMu sync.Mutex
lastBoot = make(map[string][]byte)
)
// reconnect backoff bounds; variables for testing
var (
reconnectInitialDelay = 5 * time.Second
reconnectMaxDelay = 5 * time.Minute
)
// runUpstreamSidecar dials a charger's upstream sidecar and re-dials with
// exponential backoff for as long as the charger stays connected and the rule
// remains in effect.
func runUpstreamSidecar(id string, rule ForwarderRule) {
bo := backoff.NewExponentialBackOff(
backoff.WithInitialInterval(reconnectInitialDelay),
backoff.WithMaxInterval(reconnectMaxDelay),
backoff.WithMaxElapsedTime(0),
)
for {
if !dialUpstreamSidecar(id, rule) {
delay := bo.NextBackOff()
forwarderLog.DEBUG.Printf("forwarder: reconnecting upstream for %s in %v", id, delay)
time.Sleep(delay)
continue
}
bo.Reset()
if current, ok := resolveRule(id); !ok || current.UpstreamURL == "" || !current.sameConnection(rule) {
return
}
sidecarsMu.Lock()
connected := connectedChargers[id]
_, active := sidecars[id]
sidecarsMu.Unlock()
if !connected || active {
return
}
pendingMu.Lock()
pendingMsgs[id] = nil
pendingMu.Unlock()
}
}
// resolveRule returns the forwarding rule for chargerID, or the "*" fallback.
func resolveRule(chargerID string) (ForwarderRule, bool) {
forwarderMu.RLock()
@ -261,50 +314,22 @@ func onChargerConnect(ch ws.Channel) {
pendingMsgs[id] = nil
pendingMu.Unlock()
go dialUpstreamSidecar(id, rule)
go runUpstreamSidecar(id, rule)
}
func dialUpstreamSidecar(id string, rule ForwarderRule) {
upstreamBase := strings.TrimRight(rule.UpstreamURL, "/")
upstreamPath := rule.upstreamPath(id)
var header http.Header
if rule.Username != "" || rule.Password != "" {
header = authHeader(rule.Username, rule.Password)
}
tlsConfig := &tls.Config{InsecureSkipVerify: rule.Insecure}
if rule.CaCert != "" {
caCertPool := x509.NewCertPool()
if ok := caCertPool.AppendCertsFromPEM([]byte(rule.CaCert)); !ok {
forwarderLog.WARN.Printf("forwarder: failed to parse CA cert for %s; forwarding disabled", id)
recordForwarderError(id, "invalid CA certificate")
notifyUpdated()
drainPendingWithErrors(id, nil)
return
}
tlsConfig.RootCAs = caCertPool
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
conn, _, err := websocket.Dial(ctx, upstreamBase+upstreamPath, &websocket.DialOptions{
Subprotocols: []string{"ocpp1.6"},
HTTPHeader: header,
HTTPClient: &http.Client{Transport: &http.Transport{TLSClientConfig: tlsConfig}},
})
cancel()
func dialUpstreamSidecar(id string, rule ForwarderRule) bool {
conn, err := dialUpstream(id, rule)
if err != nil {
forwarderLog.WARN.Printf("forwarder: dial upstream for %s: %v; forwarding disabled", id, err)
forwarderLog.WARN.Printf("forwarder: dial upstream for %s: %v", id, err)
recordForwarderError(id, err.Error())
notifyUpdated()
drainPendingWithErrors(id, nil)
return
return false
}
conn.SetReadLimit(-1) // no limit; OCPP frames can be large
sc := &sidecar{
chargerID: id,
upstreamURL: upstreamBase,
rule: rule,
conn: conn,
pendingUpstreamCalls: make(map[string]struct{}),
@ -314,12 +339,35 @@ func dialUpstreamSidecar(id string, rule ForwarderRule) {
// install sidecar and drain the pending buffer
sidecarsMu.Lock()
pendingMu.Lock()
old := sidecars[id]
sidecars[id] = sc
buffered := pendingMsgs[id]
delete(pendingMsgs, id)
pendingMu.Unlock()
sidecarsMu.Unlock()
// a concurrent dial may have installed a sidecar meanwhile; this one wins
if old != nil {
old.conn.CloseNow()
}
// replay the charger's BootNotification (with a fresh message id) when the
// buffer doesn't already carry one, i.e. the sidecar (re)connects mid-session;
// upstream's reply is discarded as evcc has long answered the charger
if !slices.ContainsFunc(buffered, func(frame []byte) bool {
msgType, _, action, err := parseOCPPFrame(frame)
return err == nil && msgType == ocppj.CALL && action == "BootNotification"
}) {
bootMu.Lock()
boot := lastBoot[id]
bootMu.Unlock()
if boot != nil {
if frame, err := withMessageID(boot, fmt.Sprintf("evcc-boot-%d", time.Now().UnixNano())); err == nil {
buffered = append([][]byte{frame}, buffered...)
}
}
}
// flush buffered Calls; register relay actions so upstream's reply routes back.
// CallResults/Errors are skipped: evcc already answered them.
flushed := 0
@ -333,12 +381,20 @@ func dialUpstreamSidecar(id string, rule ForwarderRule) {
sc.pendingChargerCalls[msgID] = struct{}{}
sc.pendingChargerCallsMu.Unlock()
}
if err := conn.Write(context.Background(), websocket.MessageText, frame); err != nil {
forwarderLog.ERROR.Printf("forwarder: write buffered frame to upstream for %s: %v", id, err)
writeCtx, writeCancel := context.WithTimeout(context.Background(), 10*time.Second)
writeErr := conn.Write(writeCtx, websocket.MessageText, frame)
writeCancel()
if writeErr != nil {
forwarderLog.ERROR.Printf("forwarder: write buffered frame to upstream for %s: %v", id, writeErr)
conn.CloseNow()
recordForwarderError(id, err.Error())
sidecarsMu.Lock()
if sidecars[id] == sc {
delete(sidecars, id)
}
sidecarsMu.Unlock()
recordForwarderError(id, writeErr.Error())
notifyUpdated()
return
return false
}
flushed++
}
@ -349,9 +405,11 @@ func dialUpstreamSidecar(id string, rule ForwarderRule) {
clearForwarderError(id)
notifyUpdated()
forwarderLog.INFO.Printf("forwarder: %s → %s", id, upstreamBase+upstreamPath)
forwarderLog.INFO.Printf("forwarder: %s → %s", id, strings.TrimRight(rule.UpstreamURL, "/")+rule.upstreamPath(id))
sc.readFromUpstream(rule.ReadOnly)
sc.readFromUpstream()
return true
}
// drainPendingWithErrors discards charger id's pending buffer, sending a CallError
@ -403,10 +461,13 @@ func drainPendingWithErrors(id string, sc *sidecar) {
func onChargerDisconnect(ch ws.Channel) {
id := ch.ID()
// discard pending buffer
// discard pending buffer and cached boot frame
pendingMu.Lock()
delete(pendingMsgs, id)
pendingMu.Unlock()
bootMu.Lock()
delete(lastBoot, id)
bootMu.Unlock()
sidecarsMu.Lock()
delete(connectedChargers, id)
@ -439,6 +500,13 @@ func onChargerMessage(ch ws.Channel, data []byte) bool {
return false
}
// remember the charger's boot frame for replay on sidecar (re)connect
if msgType == ocppj.CALL && action == "BootNotification" {
bootMu.Lock()
lastBoot[id] = slices.Clone(data)
bootMu.Unlock()
}
sidecarsMu.Lock()
sc := sidecars[id]
sidecarsMu.Unlock()
@ -506,7 +574,7 @@ func onChargerMessage(ch ws.Channel, data []byte) bool {
// readFromUpstream relays frames from upstream: Calls are injected into the
// charger (reply routed back), responses to bypassed charger Calls are relayed
// to the charger, others discarded. Calls are rejected in read-only mode.
func (sc *sidecar) readFromUpstream(readOnly bool) {
func (sc *sidecar) readFromUpstream() {
defer func() {
sidecarsMu.Lock()
if sidecars[sc.chargerID] == sc {
@ -541,7 +609,8 @@ func (sc *sidecar) readFromUpstream(readOnly bool) {
switch msgType {
case ocppj.CALL:
if readOnly {
// resolve the current rule so a runtime ReadOnly toggle applies live
if rule, ok := resolveRule(sc.chargerID); ok && rule.ReadOnly {
forwarderLog.DEBUG.Printf("forwarder: blocking upstream call %s in read-only session %s", msgID, sc.chargerID)
errFrame, _ := (&ocppj.CallError{
MessageTypeId: ocppj.CALL_ERROR,
@ -724,6 +793,23 @@ func extractMeterValueSampleInterval(msg []byte) (time.Duration, bool) {
return time.Duration(secs) * time.Second, true
}
// withMessageID returns a copy of a raw OCPP frame with its message id replaced.
func withMessageID(frame []byte, msgID string) ([]byte, error) {
var parts []json.RawMessage
if err := json.Unmarshal(frame, &parts); err != nil {
return nil, err
}
if len(parts) < 2 {
return nil, fmt.Errorf("invalid OCPP frame")
}
id, err := json.Marshal(msgID)
if err != nil {
return nil, err
}
parts[1] = id
return json.Marshal(parts)
}
// parseOCPPFrame extracts the message type, id and (for Calls) action from a raw frame.
func parseOCPPFrame(msg []byte) (msgType ocppj.MessageType, msgID string, action string, err error) {
var frame []json.RawMessage

174
charger/ocpp/forwarder_test.go Normal file
View file

@ -0,0 +1,174 @@
package ocpp
import (
"crypto/tls"
"net"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/coder/websocket"
"github.com/lorenzodonini/ocpp-go/ocppj"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const testBootFrame = `[2,"orig-id","BootNotification",{"chargePointVendor":"v","chargePointModel":"m"}]`
func TestForwarderWithMessageID(t *testing.T) {
frame, err := withMessageID([]byte(testBootFrame), "new-id")
require.NoError(t, err)
msgType, msgID, action, err := parseOCPPFrame(frame)
require.NoError(t, err)
assert.Equal(t, ocppj.CALL, msgType)
assert.Equal(t, "new-id", msgID)
assert.Equal(t, "BootNotification", action)
assert.Contains(t, string(frame), `"chargePointVendor":"v"`)
_, err = withMessageID([]byte(`{}`), "new-id")
assert.Error(t, err)
}
// fakeChannel implements ws.Channel for hook tests
type fakeChannel string
func (c fakeChannel) ID() string { return string(c) }
func (c fakeChannel) RemoteAddr() net.Addr { return nil }
func (c fakeChannel) TLSConnectionState() *tls.ConnectionState { return nil }
func (c fakeChannel) IsConnected() bool { return true }
func TestForwarderBootNotificationCache(t *testing.T) {
const id = "fwd-boot-cache-test"
onChargerMessage(fakeChannel(id), []byte(testBootFrame))
bootMu.Lock()
boot := lastBoot[id]
bootMu.Unlock()
assert.Equal(t, testBootFrame, string(boot))
onChargerDisconnect(fakeChannel(id))
bootMu.Lock()
boot = lastBoot[id]
bootMu.Unlock()
assert.Nil(t, boot)
}
// forwarderTestSetup marks a charger as connected with a cached boot frame and
// fast reconnect, returning a cleanup function.
func forwarderTestSetup(t *testing.T, id string) func() {
t.Helper()
delay := reconnectInitialDelay
reconnectInitialDelay = 10 * time.Millisecond
sidecarsMu.Lock()
connectedChargers[id] = true
sidecarsMu.Unlock()
bootMu.Lock()
lastBoot[id] = []byte(testBootFrame)
bootMu.Unlock()
return func() {
ApplyForwarderRules(nil)
sidecarsMu.Lock()
delete(connectedChargers, id)
sidecarsMu.Unlock()
bootMu.Lock()
delete(lastBoot, id)
bootMu.Unlock()
reconnectInitialDelay = delay
}
}
// expectBootReplay asserts that a replayed BootNotification with a fresh message id arrives.
func expectBootReplay(t *testing.T, frames <-chan []byte) {
t.Helper()
select {
case data := <-frames:
msgType, msgID, action, err := parseOCPPFrame(data)
require.NoError(t, err)
assert.Equal(t, ocppj.CALL, msgType)
assert.Equal(t, "BootNotification", action)
assert.NotEqual(t, "orig-id", msgID)
case <-time.After(5 * time.Second):
t.Fatal("timeout waiting for upstream reconnect")
}
}
func TestForwarderUpstreamReconnect(t *testing.T) {
const id = "fwd-reconnect-test"
var conns atomic.Int32
frames := make(chan []byte, 1)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{"ocpp1.6"}})
if err != nil {
return
}
if conns.Add(1) == 1 {
// simulate upstream dropping the connection without close handshake
c.CloseNow()
return
}
if _, data, err := c.Read(r.Context()); err == nil {
select {
case frames <- data:
default:
}
}
c.CloseNow()
}))
defer srv.Close()
cleanup := forwarderTestSetup(t, id)
defer cleanup()
ApplyForwarderRules([]ForwarderRule{{StationID: id, UpstreamURL: "ws" + strings.TrimPrefix(srv.URL, "http")}})
expectBootReplay(t, frames)
assert.GreaterOrEqual(t, conns.Load(), int32(2))
}
func TestForwarderDialRetry(t *testing.T) {
const id = "fwd-dial-retry-test"
var reqs atomic.Int32
frames := make(chan []byte, 1)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if reqs.Add(1) <= 2 {
// reject the websocket upgrade to simulate an unreachable upstream
http.Error(w, "unavailable", http.StatusServiceUnavailable)
return
}
c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{"ocpp1.6"}})
if err != nil {
return
}
if _, data, err := c.Read(r.Context()); err == nil {
select {
case frames <- data:
default:
}
}
c.CloseNow()
}))
defer srv.Close()
cleanup := forwarderTestSetup(t, id)
defer cleanup()
ApplyForwarderRules([]ForwarderRule{{StationID: id, UpstreamURL: "ws" + strings.TrimPrefix(srv.URL, "http")}})
expectBootReplay(t, frames)
assert.GreaterOrEqual(t, reqs.Load(), int32(3))
}

4
docs/agents/ocpp-forwarder.md
View file

@ -41,4 +41,8 @@ When a rule sets `ReadOnly`, upstream may observe but cannot control the charger
Frames that arrive from a charger before its sidecar finishes dialling are buffered (`pendingMsgs`) and flushed in order once the sidecar connects, so early messages such as `BootNotification` still reach upstream. If the dial fails or upstream drops mid-session, any buffered or in-flight relay Calls are answered to the charger with a `CallError` so it is not left hanging, and the failure is surfaced to the UI via `forwarderErrors`.
When the upstream connection fails while the charger stays connected, the sidecar is re-dialled automatically with exponential backoff (`runUpstreamSidecar`, 5s up to 5min; a successful session resets the backoff). Between attempts the relay actions fall back to evcc's local handler, so charging continues but upstream misses those transactions. The reconnect loop ends when the charger disconnects, the rule is removed, or its connection parameters change (`ApplyForwarderRules` dials its own sidecar in that case).
The last `BootNotification` of each connected charger is cached (`lastBoot`). When a sidecar connects mid-session (after an upstream reconnect or a rule added at runtime) and the pending buffer does not already carry a boot frame, the cached frame is replayed to upstream with a fresh message id, since many backends expect a boot before accepting transactions. Upstream's reply to the replay is discarded; evcc answered the charger's original boot long ago.
Rules can be changed at runtime through `ApplyForwarderRules`. Sidecars for removed rules are closed, rules with changed connection parameters are re-dialled, and rules for chargers that are not connected are test-dialled to surface unreachable hosts immediately.