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