263 lines
7.3 KiB
Go
263 lines
7.3 KiB
Go
// Package reconcile converges MediaMTX paths from the database desired state.
|
|
package reconcile
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/base64"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"time"
|
|
|
|
"yovision/sense/internal/device"
|
|
"yovision/sense/internal/metrics"
|
|
"yovision/sense/internal/onvif"
|
|
"yovision/sense/internal/store"
|
|
)
|
|
|
|
const defaultBatchSize = 128
|
|
|
|
type Repository interface {
|
|
ClaimDueReconcile(context.Context, store.ReconcileClaim) ([]store.ReconcileCandidate, error)
|
|
RenewReconcileLease(context.Context, string, string, string, time.Time, time.Duration) (bool, error)
|
|
CompleteReconcile(context.Context, string, int64, string, string, time.Time) error
|
|
FailReconcile(context.Context, string, int, time.Time, string, string, string, time.Time) error
|
|
ConvergenceSnapshot(context.Context) (store.ConvergenceSnapshot, error)
|
|
}
|
|
|
|
type MediaPaths interface {
|
|
EnsurePath(ctx context.Context, name, source string) (bool, error)
|
|
DeletePath(ctx context.Context, name string) error
|
|
}
|
|
|
|
type Reconciler struct {
|
|
repository Repository
|
|
discovery onvif.Adapter
|
|
media MediaPaths
|
|
now func() time.Time
|
|
baseBackoff time.Duration
|
|
maxBackoff time.Duration
|
|
batchSize int
|
|
instanceID string
|
|
leaseDuration time.Duration
|
|
operationTimeout time.Duration
|
|
metrics *metrics.Registry
|
|
}
|
|
|
|
type Options struct {
|
|
InstanceID string
|
|
LeaseDuration time.Duration
|
|
OperationTimeout time.Duration
|
|
Metrics *metrics.Registry
|
|
}
|
|
|
|
func New(repository Repository, discovery onvif.Adapter, media MediaPaths) *Reconciler {
|
|
return NewWithOptions(repository, discovery, media, Options{})
|
|
}
|
|
|
|
func NewWithOptions(
|
|
repository Repository,
|
|
discovery onvif.Adapter,
|
|
media MediaPaths,
|
|
options Options,
|
|
) *Reconciler {
|
|
if options.InstanceID == "" {
|
|
options.InstanceID = "single"
|
|
}
|
|
if options.LeaseDuration <= 0 {
|
|
options.LeaseDuration = 30 * time.Second
|
|
}
|
|
if options.OperationTimeout <= 0 {
|
|
options.OperationTimeout = 20 * time.Second
|
|
}
|
|
if options.OperationTimeout >= options.LeaseDuration {
|
|
options.OperationTimeout = options.LeaseDuration / 2
|
|
}
|
|
return &Reconciler{
|
|
repository: repository, discovery: discovery, media: media,
|
|
now: time.Now, baseBackoff: time.Second, maxBackoff: time.Minute, batchSize: defaultBatchSize,
|
|
instanceID: options.InstanceID, leaseDuration: options.LeaseDuration,
|
|
operationTimeout: options.OperationTimeout, metrics: options.Metrics,
|
|
}
|
|
}
|
|
|
|
func (r *Reconciler) RunOnce(ctx context.Context) (runErr error) {
|
|
started := time.Now()
|
|
defer func() {
|
|
if r.metrics != nil {
|
|
r.metrics.ObserveReconcileRun(runErr, time.Since(started))
|
|
}
|
|
}()
|
|
now := r.now().UTC()
|
|
token, err := newClaimToken()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
candidates, err := r.repository.ClaimDueReconcile(ctx, store.ReconcileClaim{
|
|
Owner: r.instanceID, Token: token, Now: now,
|
|
LeaseDuration: r.leaseDuration, Limit: r.batchSize,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("claim reconciliation candidates: %w", err)
|
|
}
|
|
var runErrors []error
|
|
for _, candidate := range candidates {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
itemNow := r.now().UTC()
|
|
renewed, err := r.repository.RenewReconcileLease(
|
|
ctx, candidate.Device.ID, r.instanceID, token, itemNow, r.leaseDuration,
|
|
)
|
|
if err != nil {
|
|
r.observeItem("error")
|
|
runErrors = append(runErrors, fmt.Errorf("renew reconcile device lease: %w", err))
|
|
continue
|
|
}
|
|
if !renewed {
|
|
r.observeItem("lease_lost")
|
|
continue
|
|
}
|
|
itemCtx, cancel := context.WithTimeout(ctx, r.operationTimeout)
|
|
err = r.reconcileOne(itemCtx, ctx, candidate, itemNow, r.instanceID, token)
|
|
cancel()
|
|
if errors.Is(err, store.ErrReconcileLeaseLost) {
|
|
r.observeItem("lease_lost")
|
|
continue
|
|
}
|
|
if err != nil {
|
|
r.observeItem("error")
|
|
runErrors = append(runErrors, fmt.Errorf("reconcile device %s: %w", candidate.Device.ID, err))
|
|
} else {
|
|
r.observeItem("success")
|
|
}
|
|
}
|
|
if r.metrics != nil {
|
|
snapshot, err := r.repository.ConvergenceSnapshot(ctx)
|
|
if err != nil {
|
|
runErrors = append(runErrors, fmt.Errorf("read convergence metrics: %w", err))
|
|
} else {
|
|
r.metrics.SetConvergence(snapshot.Total, snapshot.Unconverged)
|
|
}
|
|
}
|
|
return errors.Join(runErrors...)
|
|
}
|
|
|
|
func (r *Reconciler) reconcileOne(
|
|
ctx context.Context,
|
|
persistCtx context.Context,
|
|
candidate store.ReconcileCandidate,
|
|
now time.Time,
|
|
owner, token string,
|
|
) error {
|
|
if candidate.Device.DesiredState == device.DesiredDisabled {
|
|
if err := r.media.DeletePath(ctx, candidate.Device.PathName); err == nil {
|
|
return r.repository.CompleteReconcile(
|
|
persistCtx, candidate.Device.ID, candidate.Device.Generation, owner, token, now,
|
|
)
|
|
} else if persistCtx.Err() != nil {
|
|
return persistCtx.Err()
|
|
} else {
|
|
return r.persistFailure(persistCtx, candidate, now, owner, token, err)
|
|
}
|
|
}
|
|
result, err := r.discovery.Probe(ctx, onvif.Target{
|
|
EndpointRef: candidate.Device.EndpointRef, CredentialRef: candidate.Device.CredentialRef,
|
|
})
|
|
if err == nil {
|
|
err = validateStreamURI(result.StreamURI)
|
|
}
|
|
if err == nil {
|
|
_, err = r.media.EnsurePath(ctx, candidate.Device.PathName, result.StreamURI)
|
|
}
|
|
if err == nil {
|
|
return r.repository.CompleteReconcile(
|
|
persistCtx, candidate.Device.ID, candidate.Device.Generation, owner, token, now,
|
|
)
|
|
}
|
|
if persistCtx.Err() != nil {
|
|
return persistCtx.Err()
|
|
}
|
|
return r.persistFailure(persistCtx, candidate, now, owner, token, err)
|
|
}
|
|
|
|
func (r *Reconciler) persistFailure(
|
|
ctx context.Context,
|
|
candidate store.ReconcileCandidate,
|
|
now time.Time,
|
|
owner, token string,
|
|
err error,
|
|
) error {
|
|
failureCount := candidate.FailureCount + 1
|
|
nextAttempt := now.Add(r.backoff(failureCount))
|
|
errorCode := string(onvif.CodeOf(err))
|
|
var onvifError *onvif.Error
|
|
if !errors.As(err, &onvifError) {
|
|
errorCode = "media_error"
|
|
}
|
|
if markErr := r.repository.FailReconcile(
|
|
ctx, candidate.Device.ID, failureCount, nextAttempt, errorCode, owner, token, now,
|
|
); markErr != nil {
|
|
return errors.Join(err, fmt.Errorf("persist reconcile failure: %w", markErr))
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (r *Reconciler) observeItem(result string) {
|
|
if r.metrics != nil {
|
|
r.metrics.ObserveReconcileItem(result)
|
|
}
|
|
}
|
|
|
|
func newClaimToken() (string, error) {
|
|
value := make([]byte, 18)
|
|
if _, err := rand.Read(value); err != nil {
|
|
return "", errors.New("generate reconcile claim token")
|
|
}
|
|
return base64.RawURLEncoding.EncodeToString(value), nil
|
|
}
|
|
|
|
func validateStreamURI(value string) error {
|
|
parsed, err := url.Parse(value)
|
|
if err != nil || parsed.Host == "" || (parsed.Scheme != "rtsp" && parsed.Scheme != "rtsps") {
|
|
return &onvif.Error{Code: onvif.ErrorInvalidReply, Err: fmt.Errorf("stream URI has invalid scheme or host")}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Reconciler) backoff(failureCount int) time.Duration {
|
|
if failureCount <= 1 {
|
|
return r.baseBackoff
|
|
}
|
|
value := r.baseBackoff
|
|
for step := 1; step < failureCount; step++ {
|
|
if value >= r.maxBackoff/2 {
|
|
return r.maxBackoff
|
|
}
|
|
value *= 2
|
|
}
|
|
if value > r.maxBackoff {
|
|
return r.maxBackoff
|
|
}
|
|
return value
|
|
}
|
|
|
|
func (r *Reconciler) Run(ctx context.Context, interval time.Duration, report func(error)) {
|
|
if err := r.RunOnce(ctx); err != nil && ctx.Err() == nil && report != nil {
|
|
report(err)
|
|
}
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
if err := r.RunOnce(ctx); err != nil && ctx.Err() == nil && report != nil {
|
|
report(err)
|
|
}
|
|
}
|
|
}
|
|
}
|