From 8e8b3cc5238693151978f5eefa2b6fac63adc50a Mon Sep 17 00:00:00 2001 From: SayaAndy Date: Sun, 24 Aug 2025 21:03:10 +0700 Subject: feat: add local cache --- .gitignore | 2 + config/config-b2.sample.json | 4 +- config/config.go | 8 +- internal/client/input/b2.go | 7 +- internal/client/input/input_client_interface.go | 1 + internal/client/input/local_unix.go | 4 + main.go | 98 +++++++++++++++++++++++-- 7 files changed, 112 insertions(+), 12 deletions(-) diff --git a/.gitignore b/.gitignore index 67ec59b..186750e 100644 --- a/.gitignore +++ b/.gitignore @@ -34,3 +34,5 @@ go.work.sum config*.json !config*.sample.json + +cache.csv diff --git a/config/config-b2.sample.json b/config/config-b2.sample.json index 043c1a6..372972e 100644 --- a/config/config-b2.sample.json +++ b/config/config-b2.sample.json @@ -18,7 +18,9 @@ "jpg", "jpeg", "png" - ] + ], + "CacheProcessed": true, + "CacheProcessedCsvPath": "cache.csv" }, "Converters": [ { diff --git a/config/config.go b/config/config.go index ab051b2..1620d75 100644 --- a/config/config.go +++ b/config/config.go @@ -19,8 +19,10 @@ type Config struct { } type InputConfig struct { - Storage InputStorageConfig `json:"Storage" validate:"required"` - KnownExtensions []string `json:"KnownExtensions" validate:"required,min=0,dive,min=1"` + Storage InputStorageConfig `json:"Storage" validate:"required"` + KnownExtensions []string `json:"KnownExtensions" validate:"required,min=0,dive,min=1"` + CacheProcessed bool `json:"CacheProcessed"` + CacheProcessedCsvPath string `json:"CacheProcessedCsvPath" validate:"filepath"` } type InputStorageConfig struct { @@ -159,7 +161,7 @@ type OutputConfig struct { } type OutputLocalUnixConfig struct { - Path string `json:"Path" validate:"required,min=1"` + Path string `json:"Path" validate:"required,min=1,dirpath"` DirPermissionMode string `json:"DirPermissionMode" validate:"required,min=3"` FilePermissionMode string `json:"FilePermissionMode" validate:"required,min=3"` AttributesImplementation string `json:"AttributesImplementation" validate:"required,oneof=xattr none"` diff --git a/internal/client/input/b2.go b/internal/client/input/b2.go index db68268..fe94333 100644 --- a/internal/client/input/b2.go +++ b/internal/client/input/b2.go @@ -16,6 +16,7 @@ var _ InputClient = (*B2InputClient)(nil) type B2InputClient struct { prefix string bucket *b2.Bucket + bucketName string b2cl *b2.Client knownExtensions []string } @@ -36,7 +37,7 @@ func NewB2InputClient(cfg *config.InputConfig) (InputClient, error) { return nil, err } - return &B2InputClient{b2cl: b2cl, bucket: bucket, prefix: b2cfg.Prefix, knownExtensions: cfg.KnownExtensions}, nil + return &B2InputClient{b2cl: b2cl, bucket: bucket, bucketName: b2cfg.BucketName, prefix: b2cfg.Prefix, knownExtensions: cfg.KnownExtensions}, nil } func (c *B2InputClient) Scan() ([]string, error) { @@ -120,6 +121,10 @@ func (c *B2InputClient) ReadMetadata(path string) (*MetadataStruct, error) { return &metadata, nil } +func (c *B2InputClient) ID(path string) string { + return fmt.Sprintf("b2://%s/%s%s", c.bucketName, c.prefix, path) +} + func (c *B2InputClient) GetReader(path string) (io.ReadCloser, error) { obj := c.bucket.Object(c.prefix + path) if obj == nil { diff --git a/internal/client/input/input_client_interface.go b/internal/client/input/input_client_interface.go index 4930aa0..a2cf84b 100644 --- a/internal/client/input/input_client_interface.go +++ b/internal/client/input/input_client_interface.go @@ -11,6 +11,7 @@ type InputClient interface { Scan() ([]string, error) ReadMetadata(string) (*MetadataStruct, error) GetReader(string) (io.ReadCloser, error) + ID(path string) string } type MetadataStruct struct { diff --git a/internal/client/input/local_unix.go b/internal/client/input/local_unix.go index e509c8b..21b08a9 100644 --- a/internal/client/input/local_unix.go +++ b/internal/client/input/local_unix.go @@ -99,6 +99,10 @@ func (c *LocalUnixInputClient) ReadMetadata(path string) (*MetadataStruct, error }, nil } +func (c *LocalUnixInputClient) ID(path string) string { + return fmt.Sprintf("local-unix://%s%s", c.path, path) +} + func (c *LocalUnixInputClient) GetReader(path string) (io.ReadCloser, error) { return os.Open(c.path + path) } diff --git a/main.go b/main.go index 4994116..c3f03e7 100644 --- a/main.go +++ b/main.go @@ -2,10 +2,15 @@ package main import ( "bytes" + "encoding/csv" + "encoding/json" "flag" + "hash/crc32" + "io" "log/slog" "os" "os/signal" + "strconv" "strings" "sync" "syscall" @@ -18,6 +23,7 @@ import ( var ( configPath = flag.String("c", "config.json", "Path to the configuration file") sigTermChan = make(chan os.Signal, 1) + cacheMap = make(map[string]map[uint32]struct{}) ) func main() { @@ -48,8 +54,10 @@ func main() { } converters := make([]converter.Converter, 0, len(cfg.Converters)) - var converterTypes []string + converterTypes := make([]string, 0, len(cfg.Converters)) + converterHashes := make([]uint32, 0, len(cfg.Converters)) for _, converterCfg := range cfg.Converters { + converterBytes, _ := json.Marshal(converterCfg) conv, err := converter.NewConverterMap[converterCfg.Type](&converterCfg) if err != nil { slog.Error("fail to initialize converter", slog.String("error", err.Error())) @@ -57,6 +65,7 @@ func main() { } converters = append(converters, conv) converterTypes = append(converterTypes, converterCfg.Type) + converterHashes = append(converterHashes, crc32.ChecksumIEEE(converterBytes)) } generalLogger := slog.With(slog.String("input_storage", cfg.Input.Storage.Type)) @@ -84,6 +93,40 @@ func main() { default: } + if cfg.Input.CacheProcessed { + cacheFile, err := os.OpenFile(cfg.Input.CacheProcessedCsvPath, os.O_CREATE|os.O_RDONLY, 0644) + if err != nil { + generalLogger.Error("fail to initialize cache file", slog.String("cache_path", cfg.Input.CacheProcessedCsvPath), slog.String("error", err.Error())) + os.Exit(1) + } + defer cacheFile.Close() + + csvReader := csv.NewReader(cacheFile) + for { + rec, err := csvReader.Read() + if err == io.EOF { + break + } + if err != nil { + generalLogger.Error("fail to initialize cache while reading file", slog.String("cache_path", cfg.Input.CacheProcessedCsvPath), slog.String("error", err.Error())) + os.Exit(1) + } + if len(rec) != 2 { + generalLogger.Error("fail to initialize cache while reading file", slog.Int("record_length", len(rec)), slog.String("error", "incorrect format: expected '{image-name},{semicolon-separated-processor-hashes}'")) + os.Exit(1) + } + hashes := strings.Split(rec[1], ";") + cacheMap[rec[0]] = make(map[uint32]struct{}) + for _, hash := range hashes { + hashUint, err := strconv.ParseUint(hash, 10, 32) + if err != nil { + generalLogger.Error("fail to initialize cache while reading file", slog.Int("record_length", len(rec)), slog.String("error", err.Error())) + } + cacheMap[rec[0]][uint32(hashUint)] = struct{}{} + } + } + } + processSemaphore := make(chan struct{}, cfg.MaxProcessThreads) queueSemaphore := make(chan struct{}, cfg.MaxPreProcessThreads) var wg sync.WaitGroup @@ -106,18 +149,36 @@ func main() { default: } - inputMetadata, err := inputClient.ReadMetadata(inputName) - if err != nil { - fileLogger.Warn("fail to read metadata of (supposedly existing) input file", slog.String("error", err.Error())) - return + var inputMetadata *input.MetadataStruct + + id := inputClient.ID(file) + if _, ok := cacheMap[id]; !ok { + cacheMap[id] = make(map[uint32]struct{}) } convertersToLaunch := []int{} for j, conv := range converters { + if cfg.Input.CacheProcessed { + if _, ok := cacheMap[id][converterHashes[j]]; ok { + fileLogger.Info("skip already processed file (based on cache file containing it and processor)", + slog.String("file_id", id), + slog.Uint64("conv_hash", uint64(converterHashes[j])), + slog.Int("conv_index", j)) + continue + } + } outputName := conv.DeductOutputPath(inputName) originalInputHash := "" convLogger := fileLogger.With(slog.String("output_path", outputName), slog.Int("conv_index", j)) + if inputMetadata == nil { + inputMetadata, err = inputClient.ReadMetadata(inputName) + if err != nil { + fileLogger.Warn("fail to read metadata of (supposedly existing) input file", slog.String("error", err.Error())) + return + } + } + if !cfg.ForceRewrite && !conv.IsMissing(outputName) { outputMetadata, err := conv.ReadMetadata(outputName) if err != nil { @@ -127,6 +188,7 @@ func main() { originalInputHash = outputMetadata.HashOriginal if inputMetadata.Hash == originalInputHash { convLogger.Info("skip already processed file (based on equal hash)", slog.String("input_hash", inputMetadata.Hash)) + cacheMap[id][converterHashes[j]] = struct{}{} continue } } @@ -165,6 +227,7 @@ func main() { } convLogger.Info("successfully processed file", slog.String("input_hash", inputMetadata.Hash)) + cacheMap[id][converterHashes[convIndex]] = struct{}{} } }(i, file) } @@ -176,7 +239,28 @@ func main() { generalLogger.Info("exiting due to termination signal") os.Exit(130) default: - generalLogger.Info("all files processed successfully, exiting") - os.Exit(0) } + + generalLogger.Info("all files processed successfully") + + if cfg.Input.CacheProcessed { + generalLogger.Info("writing cache file") + cacheFile, err := os.OpenFile(cfg.Input.CacheProcessedCsvPath, os.O_TRUNC|os.O_CREATE|os.O_WRONLY, 0644) + if err != nil { + generalLogger.Error("error writing cache into file") + } else { + for id, hashes := range cacheMap { + var hashStrings = make([]string, 0, len(hashes)) + for hash := range hashes { + hashStrings = append(hashStrings, strconv.FormatUint(uint64(hash), 10)) + } + cacheFile.Write([]byte(id)) + cacheFile.Write([]byte{','}) + cacheFile.WriteString(strings.Join(hashStrings, ";")) + cacheFile.Write([]byte{'\n'}) + } + } + } + + os.Exit(0) } -- cgit v1.3.1+13