Add parquet file job archiving target

Author: moebiusband73

go.mod

@@ -1,8 +1,6 @@
 module github.com/ClusterCockpit/cc-backend
 
-go 1.24.0
-
-toolchain go1.24.1
+go 1.24.9
 
 tool (
 	github.com/99designs/gqlgen
@@ -47,6 +45,7 @@ require (
 	github.com/Azure/go-ntlmssp v0.0.0-20221128193559-754e69321358 // indirect
 	github.com/KyleBanks/depth v1.2.1 // indirect
 	github.com/agnivade/levenshtein v1.2.1 // indirect
+	github.com/andybalholm/brotli v1.1.1 // indirect
 	github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.4 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.17 // indirect
@@ -98,6 +97,10 @@ require (
 	github.com/nats-io/nkeys v0.4.12 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/oapi-codegen/runtime v1.1.1 // indirect
+	github.com/parquet-go/bitpack v1.0.0 // indirect
+	github.com/parquet-go/jsonlite v1.0.0 // indirect
+	github.com/parquet-go/parquet-go v0.27.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/common v0.67.4 // indirect
 	github.com/robfig/cron/v3 v3.0.1 // indirect
@@ -106,6 +109,7 @@ require (
 	github.com/stmcginnis/gofish v0.20.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/swaggo/files v1.0.1 // indirect
+	github.com/twpayne/go-geom v1.6.1 // indirect
 	github.com/urfave/cli/v2 v2.27.7 // indirect
 	github.com/urfave/cli/v3 v3.6.1 // indirect
 	github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 // indirect
@@ -118,6 +122,7 @@ require (
 	golang.org/x/sys v0.39.0 // indirect
 	golang.org/x/text v0.32.0 // indirect
 	golang.org/x/tools v0.40.0 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	sigs.k8s.io/yaml v1.6.0 // indirect
 )

go.sum

@@ -21,6 +21,8 @@ github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e h1:4dAU9FXIyQktp
 github.com/alexbrainman/sspi v0.0.0-20250919150558-7d374ff0d59e/go.mod h1:cEWa1LVoE5KvSD9ONXsZrj0z6KqySlCCNKHlLzbqAt4=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883 h1:bvNMNQO63//z+xNgfBlViaCIJKLlCJ6/fmUseuG0wVQ=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
+github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
+github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
 github.com/andybalholm/cascadia v1.3.3 h1:AG2YHrzJIm4BZ19iwJ/DAua6Btl3IwJX+VI4kktS1LM=
 github.com/andybalholm/cascadia v1.3.3/go.mod h1:xNd9bqTn98Ln4DwST8/nG+H0yuB8Hmgu1YHNnWw0GeA=
 github.com/antithesishq/antithesis-sdk-go v0.5.0-default-no-op h1:Ucf+QxEKMbPogRO5guBNe5cgd9uZgfoJLOYs8WWhtjM=
@@ -238,6 +240,14 @@ github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLA
 github.com/oapi-codegen/runtime v1.1.1 h1:EXLHh0DXIJnWhdRPN2w4MXAzFyE4CskzhNLUmtpMYro=
 github.com/oapi-codegen/runtime v1.1.1/go.mod h1:SK9X900oXmPWilYR5/WKPzt3Kqxn/uS/+lbpREv+eCg=
 github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
+github.com/parquet-go/bitpack v1.0.0 h1:AUqzlKzPPXf2bCdjfj4sTeacrUwsT7NlcYDMUQxPcQA=
+github.com/parquet-go/bitpack v1.0.0/go.mod h1:XnVk9TH+O40eOOmvpAVZ7K2ocQFrQwysLMnc6M/8lgs=
+github.com/parquet-go/jsonlite v1.0.0 h1:87QNdi56wOfsE5bdgas0vRzHPxfJgzrXGml1zZdd7VU=
+github.com/parquet-go/jsonlite v1.0.0/go.mod h1:nDjpkpL4EOtqs6NQugUsi0Rleq9sW/OtC1NnZEnxzF0=
+github.com/parquet-go/parquet-go v0.27.0 h1:vHWK2xaHbj+v1DYps03yDRpEsdtOeKbhiXUaixoPb3g=
+github.com/parquet-go/parquet-go v0.27.0/go.mod h1:navtkAYr2LGoJVp141oXPlO/sxLvaOe3la2JEoD8+rg=
+github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=
+github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
@@ -285,6 +295,8 @@ github.com/swaggo/http-swagger v1.3.4 h1:q7t/XLx0n15H1Q9/tk3Y9L4n210XzJF5WtnDX64
 github.com/swaggo/http-swagger v1.3.4/go.mod h1:9dAh0unqMBAlbp1uE2Uc2mQTxNMU/ha4UbucIg1MFkQ=
 github.com/swaggo/swag v1.16.6 h1:qBNcx53ZaX+M5dxVyTrgQ0PJ/ACK+NzhwcbieTt+9yI=
 github.com/swaggo/swag v1.16.6/go.mod h1:ngP2etMK5a0P3QBizic5MEwpRmluJZPHjXcMoj4Xesg=
+github.com/twpayne/go-geom v1.6.1 h1:iLE+Opv0Ihm/ABIcvQFGIiFBXd76oBIar9drAwHFhR4=
+github.com/twpayne/go-geom v1.6.1/go.mod h1:Kr+Nly6BswFsKM5sd31YaoWS5PeDDH2NftJTK7Gd028=
 github.com/urfave/cli/v2 v2.27.7 h1:bH59vdhbjLv3LAvIu6gd0usJHgoTTPhCFib8qqOwXYU=
 github.com/urfave/cli/v2 v2.27.7/go.mod h1:CyNAG/xg+iAOg0N4MPGZqVmv2rCoP267496AOXUZjA4=
 github.com/urfave/cli/v3 v3.6.1 h1:j8Qq8NyUawj/7rTYdBGrxcH7A/j7/G8Q5LhWEW4G3Mo=
@@ -293,6 +305,7 @@ github.com/vektah/gqlparser/v2 v2.5.31 h1:YhWGA1mfTjID7qJhd1+Vxhpk5HTgydrGU9IgkW
 github.com/vektah/gqlparser/v2 v2.5.31/go.mod h1:c1I28gSOVNzlfc4WuDlqU7voQnsqI6OG2amkBAFmgts=
 github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 h1:FnBeRrxr7OU4VvAzt5X7s6266i6cSVkkFPS0TuXWbIg=
 github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM=
+github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=

internal/taskmanager/retentionService.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/ClusterCockpit/cc-backend/pkg/archive"
+	pqarchive "github.com/ClusterCockpit/cc-backend/pkg/archive/parquet"
 	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
 	"github.com/go-co-op/gocron/v2"
 )
@@ -66,3 +67,96 @@ func RegisterRetentionMoveService(age int, includeDB bool, location string, omit
 				}
 			}))
 }
+
+func RegisterRetentionParquetService(retention Retention) {
+	cclog.Info("Register retention parquet service")
+
+	maxFileSizeMB := retention.MaxFileSizeMB
+	if maxFileSizeMB <= 0 {
+		maxFileSizeMB = 512
+	}
+
+	var target pqarchive.ParquetTarget
+	var err error
+
+	switch retention.TargetKind {
+	case "s3":
+		target, err = pqarchive.NewS3Target(pqarchive.S3TargetConfig{
+			Endpoint:     retention.TargetEndpoint,
+			Bucket:       retention.TargetBucket,
+			AccessKey:    retention.TargetAccessKey,
+			SecretKey:    retention.TargetSecretKey,
+			Region:       retention.TargetRegion,
+			UsePathStyle: retention.TargetUsePathStyle,
+		})
+	default:
+		target, err = pqarchive.NewFileTarget(retention.TargetPath)
+	}
+
+	if err != nil {
+		cclog.Errorf("Parquet retention: failed to create target: %v", err)
+		return
+	}
+
+	s.NewJob(gocron.DailyJob(1, gocron.NewAtTimes(gocron.NewAtTime(5, 0, 0))),
+		gocron.NewTask(
+			func() {
+				startTime := time.Now().Unix() - int64(retention.Age*24*3600)
+				jobs, err := jobRepo.FindJobsBetween(0, startTime, retention.OmitTagged)
+				if err != nil {
+					cclog.Warnf("Parquet retention: error finding jobs: %v", err)
+					return
+				}
+				if len(jobs) == 0 {
+					return
+				}
+
+				cclog.Infof("Parquet retention: processing %d jobs", len(jobs))
+				ar := archive.GetHandle()
+				pw := pqarchive.NewParquetWriter(target, maxFileSizeMB)
+
+				for _, job := range jobs {
+					meta, err := ar.LoadJobMeta(job)
+					if err != nil {
+						cclog.Warnf("Parquet retention: load meta for job %d: %v", job.JobID, err)
+						continue
+					}
+
+					data, err := ar.LoadJobData(job)
+					if err != nil {
+						cclog.Warnf("Parquet retention: load data for job %d: %v", job.JobID, err)
+						continue
+					}
+
+					row, err := pqarchive.JobToParquetRow(meta, &data)
+					if err != nil {
+						cclog.Warnf("Parquet retention: convert job %d: %v", job.JobID, err)
+						continue
+					}
+
+					if err := pw.AddJob(*row); err != nil {
+						cclog.Errorf("Parquet retention: add job %d to writer: %v", job.JobID, err)
+						continue
+					}
+				}
+
+				if err := pw.Close(); err != nil {
+					cclog.Errorf("Parquet retention: close writer: %v", err)
+					return
+				}
+
+				ar.CleanUp(jobs)
+
+				if retention.IncludeDB {
+					cnt, err := jobRepo.DeleteJobsBefore(startTime, retention.OmitTagged)
+					if err != nil {
+						cclog.Errorf("Parquet retention: delete jobs from db: %v", err)
+					} else {
+						cclog.Infof("Parquet retention: removed %d jobs from db", cnt)
+					}
+					if err = jobRepo.Optimize(); err != nil {
+						cclog.Errorf("Parquet retention: db optimization error: %v", err)
+					}
+				}
+			}))
+}

internal/taskmanager/taskManager.go

@@ -23,11 +23,20 @@ const (
 
 // Retention defines the configuration for job retention policies.
 type Retention struct {
-	Policy     string `json:"policy"`
-	Location   string `json:"location"`
-	Age        int    `json:"age"`
-	IncludeDB  bool   `json:"includeDB"`
-	OmitTagged bool   `json:"omitTagged"`
+	Policy            string `json:"policy"`
+	Location          string `json:"location"`
+	Age               int    `json:"age"`
+	IncludeDB         bool   `json:"includeDB"`
+	OmitTagged        bool   `json:"omitTagged"`
+	TargetKind        string `json:"target-kind"`
+	TargetPath        string `json:"target-path"`
+	TargetEndpoint    string `json:"target-endpoint"`
+	TargetBucket      string `json:"target-bucket"`
+	TargetAccessKey   string `json:"target-access-key"`
+	TargetSecretKey   string `json:"target-secret-key"`
+	TargetRegion      string `json:"target-region"`
+	TargetUsePathStyle bool  `json:"target-use-path-style"`
+	MaxFileSizeMB     int    `json:"max-file-size-mb"`
 }
 
 // CronFrequency defines the execution intervals for various background workers.
@@ -87,6 +96,8 @@ func initArchiveServices(config json.RawMessage) {
 			cfg.Retention.IncludeDB,
 			cfg.Retention.Location,
 			cfg.Retention.OmitTagged)
+	case "parquet":
+		RegisterRetentionParquetService(cfg.Retention)
 	}
 
 	if cfg.Compression > 0 {

pkg/archive/ConfigSchema.go

@@ -57,7 +57,7 @@ var configSchema = `
             "policy": {
               "description": "Retention policy",
               "type": "string",
-              "enum": ["none", "delete", "move"]
+              "enum": ["none", "delete", "move", "parquet"]
             },
             "include-db": {
               "description": "Also remove jobs from database",
@@ -70,6 +70,43 @@ var configSchema = `
             "location": {
               "description": "The target directory for retention. Only applicable for retention move.",
               "type": "string"
+            },
+            "target-kind": {
+              "description": "Target storage kind for parquet retention: file or s3",
+              "type": "string",
+              "enum": ["file", "s3"]
+            },
+            "target-path": {
+              "description": "Target directory path for parquet file storage",
+              "type": "string"
+            },
+            "target-endpoint": {
+              "description": "S3 endpoint URL for parquet target",
+              "type": "string"
+            },
+            "target-bucket": {
+              "description": "S3 bucket name for parquet target",
+              "type": "string"
+            },
+            "target-access-key": {
+              "description": "S3 access key for parquet target",
+              "type": "string"
+            },
+            "target-secret-key": {
+              "description": "S3 secret key for parquet target",
+              "type": "string"
+            },
+            "target-region": {
+              "description": "S3 region for parquet target",
+              "type": "string"
+            },
+            "target-use-path-style": {
+              "description": "Use path-style S3 URLs for parquet target",
+              "type": "boolean"
+            },
+            "max-file-size-mb": {
+              "description": "Maximum parquet file size in MB before splitting",
+              "type": "integer"
             }
           },
           "required": ["policy"]

pkg/archive/parquet/convert.go

@@ -0,0 +1,116 @@
+// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
+// All rights reserved. This file is part of cc-backend.
+// Use of this source code is governed by a MIT-style
+// license that can be found in the LICENSE file.
+
+package parquet
+
+import (
+	"bytes"
+	"compress/gzip"
+	"encoding/json"
+	"fmt"
+
+	"github.com/ClusterCockpit/cc-lib/v2/schema"
+)
+
+// JobToParquetRow converts job metadata and metric data into a flat ParquetJobRow.
+// Nested fields are marshaled to JSON; metric data is gzip-compressed JSON.
+func JobToParquetRow(meta *schema.Job, data *schema.JobData) (*ParquetJobRow, error) {
+	resourcesJSON, err := json.Marshal(meta.Resources)
+	if err != nil {
+		return nil, fmt.Errorf("marshal resources: %w", err)
+	}
+
+	var statisticsJSON []byte
+	if meta.Statistics != nil {
+		statisticsJSON, err = json.Marshal(meta.Statistics)
+		if err != nil {
+			return nil, fmt.Errorf("marshal statistics: %w", err)
+		}
+	}
+
+	var tagsJSON []byte
+	if len(meta.Tags) > 0 {
+		tagsJSON, err = json.Marshal(meta.Tags)
+		if err != nil {
+			return nil, fmt.Errorf("marshal tags: %w", err)
+		}
+	}
+
+	var metaDataJSON []byte
+	if meta.MetaData != nil {
+		metaDataJSON, err = json.Marshal(meta.MetaData)
+		if err != nil {
+			return nil, fmt.Errorf("marshal metadata: %w", err)
+		}
+	}
+
+	var footprintJSON []byte
+	if meta.Footprint != nil {
+		footprintJSON, err = json.Marshal(meta.Footprint)
+		if err != nil {
+			return nil, fmt.Errorf("marshal footprint: %w", err)
+		}
+	}
+
+	var energyFootJSON []byte
+	if meta.EnergyFootprint != nil {
+		energyFootJSON, err = json.Marshal(meta.EnergyFootprint)
+		if err != nil {
+			return nil, fmt.Errorf("marshal energy footprint: %w", err)
+		}
+	}
+
+	metricDataGz, err := compressJobData(data)
+	if err != nil {
+		return nil, fmt.Errorf("compress metric data: %w", err)
+	}
+
+	return &ParquetJobRow{
+		JobID:          meta.JobID,
+		Cluster:        meta.Cluster,
+		SubCluster:     meta.SubCluster,
+		Partition:      meta.Partition,
+		Project:        meta.Project,
+		User:           meta.User,
+		State:          string(meta.State),
+		StartTime:      meta.StartTime,
+		Duration:       meta.Duration,
+		Walltime:       meta.Walltime,
+		NumNodes:       meta.NumNodes,
+		NumHWThreads:   meta.NumHWThreads,
+		NumAcc:         meta.NumAcc,
+		Exclusive:      meta.Exclusive,
+		Energy:         meta.Energy,
+		SMT:            meta.SMT,
+		ResourcesJSON:  resourcesJSON,
+		StatisticsJSON: statisticsJSON,
+		TagsJSON:       tagsJSON,
+		MetaDataJSON:   metaDataJSON,
+		FootprintJSON:  footprintJSON,
+		EnergyFootJSON: energyFootJSON,
+		MetricDataGz:   metricDataGz,
+	}, nil
+}
+
+func compressJobData(data *schema.JobData) ([]byte, error) {
+	jsonBytes, err := json.Marshal(data)
+	if err != nil {
+		return nil, err
+	}
+
+	var buf bytes.Buffer
+	gz, err := gzip.NewWriterLevel(&buf, gzip.BestCompression)
+	if err != nil {
+		return nil, err
+	}
+	if _, err := gz.Write(jsonBytes); err != nil {
+		return nil, err
+	}
+	if err := gz.Close(); err != nil {
+		return nil, err
+	}
+
+	return buf.Bytes(), nil
+}