gustav/internal/pipeline/reader.go
2024-09-08 01:40:03 +03:00

74 lines
1.4 KiB
Go

package pipeline
import (
"context"
"fmt"
"io"
"io/fs"
"os"
"path/filepath"
"runtime"
"strings"
"golang.org/x/sync/errgroup"
"gopkg.in/yaml.v3"
)
func FindPipelines(path string) []string {
_, b, _, _ := runtime.Caller(0)
root := filepath.Join(filepath.Dir(b), "../..")
pipelinesPaths := []string{}
filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
if err != nil {
return err
}
if strings.Contains(d.Name(), "pipeline.yaml") || strings.Contains(d.Name(), "pipeline.yml") {
pipelinesPaths = append(pipelinesPaths, path)
}
return nil
})
return pipelinesPaths
}
func LoadPipelines(ctx context.Context, paths []string) ([]*Pipeline, error) {
pipelines := make([]*Pipeline, len(paths))
eg, ctx := errgroup.WithContext(ctx)
for i, p := range paths {
i := i
eg.Go(func() error {
f, err := os.Open(p)
if err != nil {
return fmt.Errorf("failed to read pipeline file (%s): %w", p, err)
}
data, err := io.ReadAll(f)
if err != nil {
return fmt.Errorf("failed to read pipeline config (%s): %w", p, err)
}
pipeline := new(Pipeline)
if err := yaml.Unmarshal(data, pipeline); err != nil {
return fmt.Errorf("failed to unmarshall pipeline file (%s): %w", p, err)
}
pipelines[i] = pipeline
return nil
})
}
if err := eg.Wait(); err != nil {
return nil, err
}
return pipelines, nil
}