From ed916eb0424d4375b998bd9b206d5604552114e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?C=C3=A9dric=20Verstraeten?= Date: Wed, 5 Aug 2026 15:12:51 +0200 Subject: [PATCH] Extend recording upload metadata with duration and timestamp The upload marker now carries filename, device key, timestamp and duration alongside FPS, populated from the finalized MP4 at recording time. Uploads propagate the new fields: legacy uploads add X-Kerberos-Storage-Duration and X-Kerberos-Storage-Timestamp headers, and resumable (tus) uploads include duration and timestamp in Upload-Metadata. setQueuedRecordingFPSHeader is renamed to setQueuedRecordingMetadataHeaders, and decoding of historical markers remains backwards compatible. --- machinery/src/capture/main.go | 33 +++++++++++-------- machinery/src/capture/main_test.go | 28 ++++++++++++---- machinery/src/cloud/kerberos_hub.go | 4 +-- machinery/src/cloud/kerberos_vault.go | 2 +- machinery/src/cloud/recording_metadata.go | 32 ++++++++++++++++-- machinery/src/cloud/tus_client.go | 25 +++++++++++--- machinery/src/cloud/tus_client_test.go | 33 ++++++++++++++++--- .../src/models/recording_upload_metadata.go | 6 +++- 8 files changed, 128 insertions(+), 35 deletions(-) diff --git a/machinery/src/capture/main.go b/machinery/src/capture/main.go index f9f7e38..87a4aad 100644 --- a/machinery/src/capture/main.go +++ b/machinery/src/capture/main.go @@ -53,16 +53,23 @@ func publishRecordingState(mqttClient mqtt.Client, hubKey string, configuration } } -// queueRecordingForUpload creates the marker consumed by the upload worker and -// stores the average FPS of the finalized recording in it. Empty markers remain -// valid for recordings whose FPS cannot be determined. -func queueRecordingForUpload(configDirectory, name string, value float64) { - metadata := models.RecordingUploadMetadata{} - if value > 0 && value <= 240 && !math.IsInf(value, 0) && !math.IsNaN(value) { - if rounded := int(math.Floor(value)); rounded > 0 { - metadata.FPS = rounded - } +func recordingUploadMetadata(name, deviceKey string, timestamp int64, mp4Video *video.MP4) models.RecordingUploadMetadata { + metadata := models.RecordingUploadMetadata{ + FileName: filepath.Base(name), + DeviceKey: deviceKey, + Timestamp: timestamp, + Duration: mp4Video.VideoTotalDuration, } + value := mp4Video.AverageFPS() + if value > 0 && value <= 240 && !math.IsInf(value, 0) && !math.IsNaN(value) { + metadata.FPS = int(math.Floor(value)) + } + return metadata +} + +// queueRecordingForUpload creates the marker consumed by the upload worker and +// stores metadata captured from the finalized recording. +func queueRecordingForUpload(configDirectory string, metadata models.RecordingUploadMetadata) { payload, err := json.Marshal(metadata) if err != nil { log.Log.Error("capture.main.queueRecordingForUpload(): " + err.Error()) @@ -85,7 +92,7 @@ func queueRecordingForUpload(configDirectory, name string, value float64) { defer os.Remove(marker.Name()) } if err == nil { - err = os.Rename(marker.Name(), filepath.Join(configDirectory, "data", "cloud", models.RecordingUploadMetadataFileName(name))) + err = os.Rename(marker.Name(), filepath.Join(configDirectory, "data", "cloud", models.RecordingUploadMetadataFileName(metadata.FileName))) } if err != nil { log.Log.Error("capture.main.queueRecordingForUpload(): " + err.Error()) @@ -495,7 +502,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat } } - queueRecordingForUpload(configDirectory, name, mp4Video.AverageFPS()) + queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video)) recordingStatus = "idle" @@ -652,7 +659,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat } } - queueRecordingForUpload(configDirectory, name, mp4Video.AverageFPS()) + queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, startRecording, mp4Video)) recordingStatus = "idle" @@ -922,7 +929,7 @@ func HandleRecordStream(queue *packets.Queue, configDirectory string, configurat } } - queueRecordingForUpload(configDirectory, name, mp4Video.AverageFPS()) + queueRecordingForUpload(configDirectory, recordingUploadMetadata(name, config.Key, displayTime, mp4Video)) // Clean up the recording directory if necessary. CleanupRecordingDirectory(configDirectory, configuration) diff --git a/machinery/src/capture/main_test.go b/machinery/src/capture/main_test.go index b771eb3..72d6f53 100644 --- a/machinery/src/capture/main_test.go +++ b/machinery/src/capture/main_test.go @@ -1,26 +1,36 @@ package capture import ( + "encoding/json" "math" "os" "path/filepath" "testing" + + "github.com/kerberos-io/agent/machinery/src/models" + "github.com/kerberos-io/agent/machinery/src/video" ) -func TestQueueRecordingForUploadStoresFinalizedFPS(t *testing.T) { +func TestQueueRecordingForUploadStoresFinalizedMetadata(t *testing.T) { configDirectory := t.TempDir() if err := os.MkdirAll(filepath.Join(configDirectory, "data", "cloud"), 0o755); err != nil { t.Fatalf("mkdir cloud queue: %v", err) } - queueRecordingForUpload(configDirectory, "recording.mp4", 29.970029) + mp4Video := &video.MP4{VideoTotalDuration: 20452, SampleCount: 613} + metadata := recordingUploadMetadata("recording.mp4", "device-key", 1785934709414, mp4Video) + queueRecordingForUpload(configDirectory, metadata) got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.metadata")) if err != nil { t.Fatalf("read upload marker: %v", err) } - if string(got) != `{"fps":29}` { - t.Fatalf("upload marker = %q, want JSON metadata", got) + var stored models.RecordingUploadMetadata + if err := json.Unmarshal(got, &stored); err != nil { + t.Fatalf("decode upload marker: %v", err) + } + if stored.FileName != "recording.mp4" || stored.DeviceKey != "device-key" || stored.Timestamp != 1785934709414 || stored.Duration != 20452 || stored.FPS != 29 { + t.Fatalf("upload marker = %+v", stored) } } @@ -32,14 +42,18 @@ func TestQueueRecordingForUploadKeepsUnknownFPSCompatible(t *testing.T) { t.Fatalf("mkdir cloud queue: %v", err) } - queueRecordingForUpload(configDirectory, "recording.mp4", fps) + metadata := models.RecordingUploadMetadata{FileName: "recording.mp4"} + if fps >= 1 && fps <= 240 && !math.IsNaN(fps) && !math.IsInf(fps, 0) { + metadata.FPS = int(math.Floor(fps)) + } + queueRecordingForUpload(configDirectory, metadata) got, err := os.ReadFile(filepath.Join(configDirectory, "data", "cloud", "recording.metadata")) if err != nil { t.Fatalf("read upload marker: %v", err) } - if string(got) != `{}` { - t.Fatalf("upload marker = %q, want empty JSON object", got) + if string(got) != `{"filename":"recording.mp4","device_key":"","timestamp":0,"duration":0}` { + t.Fatalf("upload marker = %q, want metadata without FPS", got) } }) } diff --git a/machinery/src/cloud/kerberos_hub.go b/machinery/src/cloud/kerberos_hub.go index abfc6ee..26809ac 100644 --- a/machinery/src/cloud/kerberos_hub.go +++ b/machinery/src/cloud/kerberos_hub.go @@ -85,7 +85,7 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo req.Header.Set("X-Kerberos-Hub-PublicKey", config.HubKey) req.Header.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey) req.Header.Set("X-Kerberos-Hub-Region", config.S3.Region) - setQueuedRecordingFPSHeader(req.Header, fileName) + setQueuedRecordingMetadataHeaders(req.Header, fileName) var client *http.Client if os.Getenv("AGENT_TLS_INSECURE") == "true" { @@ -129,7 +129,7 @@ func UploadKerberosHub(configuration *models.Configuration, fileName string) (bo req.Header.Set("X-Kerberos-Hub-PublicKey", config.HubKey) req.Header.Set("X-Kerberos-Hub-PrivateKey", config.HubPrivateKey) req.Header.Set("X-Kerberos-Hub-Region", config.S3.Region) - setQueuedRecordingFPSHeader(req.Header, fileName) + setQueuedRecordingMetadataHeaders(req.Header, fileName) resp, err = client.Do(req) if resp != nil { defer resp.Body.Close() diff --git a/machinery/src/cloud/kerberos_vault.go b/machinery/src/cloud/kerberos_vault.go index 4f2a3ea..861fdce 100644 --- a/machinery/src/cloud/kerberos_vault.go +++ b/machinery/src/cloud/kerberos_vault.go @@ -165,7 +165,7 @@ func uploadVaultLegacy(vault models.KStorage, publicKey, deviceKey, fileName, la } req.Header.Set("Content-Type", "video/mp4") setVaultHeaders(req.Header, vault, publicKey, deviceKey, fileName) - setQueuedRecordingFPSHeader(req.Header, fileName) + setQueuedRecordingMetadataHeaders(req.Header, fileName) client := newVaultHTTPClient(0) resp, err := client.Do(req) diff --git a/machinery/src/cloud/recording_metadata.go b/machinery/src/cloud/recording_metadata.go index d23c6ff..50c5979 100644 --- a/machinery/src/cloud/recording_metadata.go +++ b/machinery/src/cloud/recording_metadata.go @@ -13,6 +13,8 @@ import ( ) const recordingFPSHeader = "X-Kerberos-Storage-Fps" +const recordingDurationHeader = "X-Kerberos-Storage-Duration" +const recordingTimestampHeader = "X-Kerberos-Storage-Timestamp" // queuedRecordingFPS reads the FPS snapshot written into the upload marker // when the recording was finalized. Historical empty markers intentionally @@ -25,8 +27,8 @@ func queuedRecordingFPS(fileName string) string { marker := strings.TrimSpace(string(value)) if strings.HasPrefix(marker, "{") { - var metadata models.RecordingUploadMetadata - if err := json.Unmarshal(value, &metadata); err != nil || metadata.FPS <= 0 || metadata.FPS > 240 { + metadata, ok := decodeRecordingUploadMetadata(value) + if !ok || metadata.FPS <= 0 || metadata.FPS > 240 { return "" } return strconv.Itoa(metadata.FPS) @@ -41,6 +43,22 @@ func queuedRecordingFPS(fileName string) string { return fps } +func queuedRecordingMetadata(fileName string) (models.RecordingUploadMetadata, bool) { + value, ok := readRecordingUploadMetadata(fileName) + if !ok || !strings.HasPrefix(strings.TrimSpace(string(value)), "{") { + return models.RecordingUploadMetadata{}, false + } + return decodeRecordingUploadMetadata(value) +} + +func decodeRecordingUploadMetadata(value []byte) (models.RecordingUploadMetadata, bool) { + var metadata models.RecordingUploadMetadata + if err := json.Unmarshal(value, &metadata); err != nil { + return models.RecordingUploadMetadata{}, false + } + return metadata, true +} + func readRecordingUploadMetadata(fileName string) ([]byte, bool) { markerNames := []string{ models.RecordingUploadMetadataFileName(fileName), @@ -55,8 +73,16 @@ func readRecordingUploadMetadata(fileName string) ([]byte, bool) { return nil, false } -func setQueuedRecordingFPSHeader(header http.Header, fileName string) { +func setQueuedRecordingMetadataHeaders(header http.Header, fileName string) { if fps := queuedRecordingFPS(fileName); fps != "" { header.Set(recordingFPSHeader, fps) } + if metadata, ok := queuedRecordingMetadata(fileName); ok { + if metadata.Duration > 0 { + header.Set(recordingDurationHeader, strconv.FormatUint(metadata.Duration, 10)) + } + if metadata.Timestamp > 0 { + header.Set(recordingTimestampHeader, strconv.FormatInt(metadata.Timestamp, 10)) + } + } } diff --git a/machinery/src/cloud/tus_client.go b/machinery/src/cloud/tus_client.go index e40c290..d6ce595 100644 --- a/machinery/src/cloud/tus_client.go +++ b/machinery/src/cloud/tus_client.go @@ -305,7 +305,7 @@ func runTusUpload(baseURL, metadata, fileName, label, slot string, setHeaders tu // is additionally carried in the tus Upload-Metadata. func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, label, slot string) (bool, bool, bool, string, error) { baseURL := strings.TrimRight(vault.URI, "/") + tusUploadPath - metadata := encodeTusMetadata(map[string]string{ + metadataValues := map[string]string{ "filename": fileName, "device": deviceKey, "directory": vault.Directory, @@ -313,7 +313,9 @@ func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, "capture": "IPCamera", "cloudkey": publicKey, "fps": queuedRecordingFPS(fileName), - }) + } + addRecordingTusMetadata(metadataValues, fileName) + metadata := encodeTusMetadata(metadataValues) setHeaders := func(h http.Header, fn string) { setVaultTusHeaders(h, vault, publicKey, deviceKey, fn) } @@ -327,18 +329,33 @@ func uploadVaultResumable(vault models.KStorage, publicKey, deviceKey, fileName, // intentionally omitted from the metadata here. func uploadHubResumable(config *models.Config, fileName, label, slot string) (bool, bool, bool, string, error) { baseURL := strings.TrimRight(config.HubURI, "/") + tusUploadPath - metadata := encodeTusMetadata(map[string]string{ + metadataValues := map[string]string{ "filename": fileName, "device": config.Key, "capture": "IPCamera", "fps": queuedRecordingFPS(fileName), - }) + } + addRecordingTusMetadata(metadataValues, fileName) + metadata := encodeTusMetadata(metadataValues) setHeaders := func(h http.Header, fn string) { setHubTusHeaders(h, config, fn) } return runTusUpload(baseURL, metadata, fileName, label, slot, setHeaders) } +func addRecordingTusMetadata(values map[string]string, fileName string) { + metadata, ok := queuedRecordingMetadata(fileName) + if !ok { + return + } + if metadata.Duration > 0 { + values["duration"] = strconv.FormatUint(metadata.Duration, 10) + } + if metadata.Timestamp > 0 { + values["timestamp"] = strconv.FormatInt(metadata.Timestamp, 10) + } +} + // tusCreate performs the tus "creation" request (POST). On success it returns // the resolved upload URL the agent should use for subsequent HEAD/PATCH calls. func tusCreate(client *http.Client, baseURL string, size int64, metadata string, setHeaders tusHeaderFunc, fileName string) (string, int, error) { diff --git a/machinery/src/cloud/tus_client_test.go b/machinery/src/cloud/tus_client_test.go index 6077993..81a2c5e 100644 --- a/machinery/src/cloud/tus_client_test.go +++ b/machinery/src/cloud/tus_client_test.go @@ -272,7 +272,7 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) { fileName := "1564859471_6-474162_oprit_577-283-727-375_1153_27.mp4" payload := bytes.Repeat([]byte("x"), 4096) withRecording(t, fileName, payload) - withQueuedRecordingFPS(t, fileName, `{"fps":29}`) + withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":29}`) uploaded, responded, supported, _, err := uploadVaultResumable(testVault(ts.URL), "pk", "dev", fileName, "test", "primary") if err != nil { @@ -288,9 +288,16 @@ func TestUploadVaultResumable_HappyPath(t *testing.T) { t.Fatalf("expected sidecar to be removed after success, stat err = %v", err) } posts := srv.requestsForMethod(http.MethodPost) - if got := decodeTusMetadata(posts[0].header.Get("Upload-Metadata"))["fps"]; got != "29" { + metadata := decodeTusMetadata(posts[0].header.Get("Upload-Metadata")) + if got := metadata["fps"]; got != "29" { t.Fatalf("POST metadata fps = %q, want %q", got, "29") } + if got := metadata["duration"]; got != "20452" { + t.Fatalf("POST metadata duration = %q, want %q", got, "20452") + } + if got := metadata["timestamp"]; got != "1785934709414" { + t.Fatalf("POST metadata timestamp = %q, want %q", got, "1785934709414") + } } func TestQueuedRecordingFPSValidation(t *testing.T) { @@ -323,7 +330,7 @@ func TestQueuedRecordingFPSValidation(t *testing.T) { } header := make(http.Header) - setQueuedRecordingFPSHeader(header, fileName) + setQueuedRecordingMetadataHeaders(header, fileName) if got := header.Get(recordingFPSHeader); got != test.want { t.Fatalf("legacy FPS header = %q, want %q", got, test.want) } @@ -339,12 +346,30 @@ func TestQueuedRecordingFPSAllowsMissingHistoricalMarker(t *testing.T) { t.Fatalf("queuedRecordingFPS() = %q, want empty for missing marker", got) } header := make(http.Header) - setQueuedRecordingFPSHeader(header, fileName) + setQueuedRecordingMetadataHeaders(header, fileName) if got := header.Get(recordingFPSHeader); got != "" { t.Fatalf("legacy FPS header = %q, want empty for missing marker", got) } } +func TestQueuedRecordingMetadataHeaders(t *testing.T) { + fileName := "recording.mp4" + withRecording(t, fileName, []byte("recording")) + withQueuedRecordingFPS(t, fileName, `{"filename":"recording.mp4","device_key":"device-key","timestamp":1785934709414,"duration":20452,"fps":25}`) + + header := make(http.Header) + setQueuedRecordingMetadataHeaders(header, fileName) + if got := header.Get(recordingFPSHeader); got != "25" { + t.Fatalf("FPS header = %q", got) + } + if got := header.Get(recordingDurationHeader); got != "20452" { + t.Fatalf("duration header = %q", got) + } + if got := header.Get(recordingTimestampHeader); got != "1785934709414" { + t.Fatalf("timestamp header = %q", got) + } +} + func TestQueuedRecordingFPSAllowsLegacyMarkerFileName(t *testing.T) { fileName := "recording.mp4" withRecording(t, fileName, []byte("recording")) diff --git a/machinery/src/models/recording_upload_metadata.go b/machinery/src/models/recording_upload_metadata.go index 243237d..5975e6b 100644 --- a/machinery/src/models/recording_upload_metadata.go +++ b/machinery/src/models/recording_upload_metadata.go @@ -11,7 +11,11 @@ const RecordingUploadMetadataExtension = ".metadata" // with a recording. New optional fields can be added without changing the queue // mechanism or breaking older agents. type RecordingUploadMetadata struct { - FPS int `json:"fps,omitempty"` + FileName string `json:"filename"` + DeviceKey string `json:"device_key"` + Timestamp int64 `json:"timestamp"` // Unix milliseconds. + Duration uint64 `json:"duration"` // Milliseconds. + FPS int `json:"fps,omitempty"` } // RecordingUploadMetadataFileName returns the queue marker name associated