diff --git a/.forgejo/workflows/default.yml b/.forgejo/workflows/default.yml new file mode 100644 index 0000000..8212505 --- /dev/null +++ b/.forgejo/workflows/default.yml @@ -0,0 +1,21 @@ +name: default + +on: + push: + branches: [main] + +jobs: + default: + name: dotnet-default-workflow + runs-on: debian-latest + container: docker.io/golang:1.24-alpine + steps: + - name: checkout + uses: https://code.geekeey.de/actions/checkout@1 + - name: go run + run: | + export INPUT_NAME=foo + export INPUT_PATH=. + export INPUT_PATTERN=**/* + if=/dev/urandom of=bigfile_rand.bin bs=1M count=10 + go run push/main.go diff --git a/artifacts.go b/artifacts.go index 3babd4d..a3ca478 100644 --- a/artifacts.go +++ b/artifacts.go @@ -38,8 +38,8 @@ func JobInfo() (string, string, error) { } type jwt struct { - Header map[string]interface{} - Payload map[string]interface{} + Header map[string]any + Payload map[string]any Signature string } @@ -60,7 +60,7 @@ func newJwtDecode(token string) (*jwt, error) { if err != nil { return nil, fmt.Errorf("invalid JWT header: %v", err) } - var header map[string]interface{} + var header map[string]any if err := json.Unmarshal(headerBytes, &header); err != nil { return nil, fmt.Errorf("invalid JWT header json: %v", err) } @@ -69,7 +69,7 @@ func newJwtDecode(token string) (*jwt, error) { if err != nil { return nil, fmt.Errorf("invalid JWT payload: %v", err) } - var payload map[string]interface{} + var payload map[string]any if err := json.Unmarshal(payloadBytes, &payload); err != nil { return nil, fmt.Errorf("invalid JWT payload json: %v", err) } diff --git a/client.go b/client.go index c897ebe..936a14f 100644 --- a/client.go +++ b/client.go @@ -80,7 +80,7 @@ type CreateArtifactRequest struct { type CreateArtifactResponse struct { Ok bool `json:"ok"` - SignedUploadUrl string `json:"signed_upload_url"` + SignedUploadUrl string `json:"signedUploadUrl"` } type FinalizeArtifactRequest struct { @@ -99,8 +99,8 @@ type FinalizeArtifactResponse struct { type ListArtifactsRequest struct { RunID string `json:"workflowRunBackendId"` JobRunID string `json:"workflowJobRunBackendId"` - NameFilter string `json:nameFilter` - IdFilter string `json:idFilter` + NameFilter string `json:"nameFilter"` + IdFilter string `json:"idFilter"` } type ListArtifactsResponse struct { @@ -112,7 +112,7 @@ type ListArtifactsResponse struct { Size string `json:"size"` CreatedAt *time.Time `json:"createdAt"` Digest string `json:"digest"` - } `json:artifacts` + } `json:"artifacts"` } type GetSignedArtifactURLRequest struct { @@ -166,7 +166,7 @@ func (gh *GitHub) twirp(ctx context.Context, url string, in any, out any) error return err } req.Header.Set("content-type", "application/json") - req.Header.Set("Accept", "application/json") + req.Header.Set("accept", "application/json") res, err := gh.client.Do(req) if err != nil { diff --git a/go.mod b/go.mod index 3daec7b..ca10462 100644 --- a/go.mod +++ b/go.mod @@ -2,4 +2,4 @@ module code.geekeey.de/actions/artifacts go 1.23.2 -require code.geekeey.de/actions/sdk v1.2.0 +require code.geekeey.de/actions/sdk v1.4.0 diff --git a/go.sum b/go.sum index 6a21fc7..34a8f74 100644 --- a/go.sum +++ b/go.sum @@ -1,2 +1,6 @@ code.geekeey.de/actions/sdk v1.2.0 h1:qQvOdTBtZVpR1xATNnApeiW78y34Jh83OjsQg1ajPjA= code.geekeey.de/actions/sdk v1.2.0/go.mod h1:7iIOi3YUmdDe2BCz9s59wK1t0Ckec34e664q+TOreiE= +code.geekeey.de/actions/sdk v1.3.0 h1:F5o5cHG829XeLv/Ec0PiUP6kX4smyL4Xn4vGl0qvp+0= +code.geekeey.de/actions/sdk v1.3.0/go.mod h1:7iIOi3YUmdDe2BCz9s59wK1t0Ckec34e664q+TOreiE= +code.geekeey.de/actions/sdk v1.4.0 h1:u97FwLq80xLpTvOun3SrNSGhX/WlAI52mP1eyjpIw2U= +code.geekeey.de/actions/sdk v1.4.0/go.mod h1:7iIOi3YUmdDe2BCz9s59wK1t0Ckec34e664q+TOreiE= diff --git a/load/main.go b/pull/main.go similarity index 63% rename from load/main.go rename to pull/main.go index b80b539..00a783f 100644 --- a/load/main.go +++ b/pull/main.go @@ -9,7 +9,7 @@ import ( ) func main() { - action := &SaveArtifactAction{Action: sdk.New()} + action := &pullArtifactAction{Action: sdk.New()} ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) defer cancel() if err := action.Run(ctx); err != nil { @@ -18,15 +18,15 @@ func main() { } } -type SaveArtifactAction struct { +type pullArtifactAction struct { *sdk.Action name string } -func (action *SaveArtifactAction) setup() error { +func (action *pullArtifactAction) setup() error { return nil } -func (action *SaveArtifactAction) Run(ctx context.Context) error { +func (action *pullArtifactAction) Run(ctx context.Context) error { return nil } diff --git a/push/main.go b/push/main.go new file mode 100644 index 0000000..6dbea63 --- /dev/null +++ b/push/main.go @@ -0,0 +1,142 @@ +package main + +import ( + "archive/zip" + "compress/flate" + "context" + "fmt" + "io" + "os" + "os/signal" + "runtime" + "strings" + "time" + + "code.geekeey.de/actions/artifacts" + "code.geekeey.de/actions/sdk" + "code.geekeey.de/actions/sdk/glob" +) + +func main() { + action := &pushArtifactAction{Action: sdk.New()} + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) + defer cancel() + if err := action.Run(ctx); err != nil { + action.Errorf("%s", err) + os.Exit(1) + } +} + +type pushArtifactAction struct { + *sdk.Action + name string + path string + patterns []string +} + +func (action *pushArtifactAction) setup() error { + action.name = action.GetInput("name") + if len(action.name) == 0 { + return fmt.Errorf("input 'name': is empty") + } + + action.path = action.GetInput("path") + if len(action.path) == 0 { + return fmt.Errorf("input 'path': is empty") + } + + patterns := strings.Split(action.GetInput("pattern"), "\n") + for _, pattern := range patterns { + if len(pattern) == 0 { + continue + } + action.patterns = append(action.patterns, pattern) + } + return nil +} + +func (action *pushArtifactAction) Run(ctx context.Context) error { + if err := action.setup(); err != nil { + return err + } + + run, job, err := artifacts.JobInfo() + if err != nil { + return fmt.Errorf("unable to get job info: %v", err) + } + + github := artifacts.NewGitHub(artifacts.NewActionClient()) + + expire := time.Now().Add(20 * time.Hour) + create, err := github.CreateArtifact(ctx, artifacts.CreateArtifactRequest{ + RunID: run, + JobRunID: job, + Name: action.name, + Version: 4, + ExpiresAt: &expire, + }) + if err != nil { + return fmt.Errorf("cannot create artifact: %w", err) + } + if !create.Ok { + return fmt.Errorf("cannot get pre-signed URL") + } + + rd, err := action.CreateArtifact(action.path, action.patterns, ctx) + if err != nil { + return fmt.Errorf("cannot create archive: %w", err) + } + + res, err := artifacts.PushBlob(ctx, rd, create.SignedUploadUrl, 1024*1024*1, runtime.NumCPU()) + if err != nil { + return fmt.Errorf("cannot upload artifact: %w", err) + } + + finish, err := github.FinalizeArtifact(ctx, artifacts.FinalizeArtifactRequest{ + RunID: run, + JobRunID: job, + Name: action.name, + Size: res.Size, + Hash: res.SHA256Sum, + }) + if err != nil { + return fmt.Errorf("cannot finish artifact: %w", err) + } + if !create.Ok { + return fmt.Errorf("cannot finish artifact upload") + } + + action.Noticef("created artifact: %s", finish.ArtifactId) + + return nil +} + +func (action *pushArtifactAction) CreateArtifact(path string, inputs []string, ctx context.Context) (io.Reader, error) { + if stat, err := os.Stat(path); err != nil || !stat.IsDir() { + if err != nil { + return nil, fmt.Errorf("cannot find directory: %w", err) + } else { + return nil, fmt.Errorf("source path is not a directory") + } + } + + gfs, err := glob.NewGlobFS(os.DirFS(path), inputs...) + if err != nil { + return nil, fmt.Errorf("cannot apply glob patterns: %w", err) + } + + rd, wr, err := os.Pipe() + go func() { + defer wr.Close() + + a := zip.NewWriter(wr) + a.RegisterCompressor(zip.Deflate, func(w io.Writer) (io.WriteCloser, error) { return flate.NewWriter(w, flate.BestCompression) }) + if err := a.AddFS(gfs); err != nil { + panic(err) + } + if err := a.Close(); err != nil { + panic(err) + } + }() + return rd, err +} diff --git a/save/main.go b/save/main.go deleted file mode 100644 index 7693ffc..0000000 --- a/save/main.go +++ /dev/null @@ -1,96 +0,0 @@ -package main - -import ( - "archive/tar" - "context" - "fmt" - "io" - "os" - "os/signal" - "runtime" - "time" - - "code.geekeey.de/actions/artifacts" - "code.geekeey.de/actions/sdk" - "code.geekeey.de/actions/sdk/iox" -) - -func main() { - action := &SaveArtifactAction{Action: sdk.New()} - ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt) - defer cancel() - if err := action.Run(ctx); err != nil { - action.Errorf("%s", err) - os.Exit(1) - } -} - -type SaveArtifactAction struct { - *sdk.Action - name string -} - -func (action *SaveArtifactAction) setup() error { - return nil -} - -func (action *SaveArtifactAction) Run(ctx context.Context) error { - if err := action.setup(); err != nil { - return err - } - - run, job, err := artifacts.JobInfo() - if err != nil { - return fmt.Errorf("unable to get job info: %v", err) - } - - github := artifacts.NewGitHub(artifacts.NewActionClient()) - - expire := time.Now().Add(20 * time.Hour) - create, err := github.CreateArtifact(ctx, artifacts.CreateArtifactRequest{ - RunID: run, - JobRunID: job, - Name: action.name, - Version: 4, - ExpiresAt: &expire, - }) - if err != nil { - return fmt.Errorf("cannot create artifact %w", err) - } - if !create.Ok { - return fmt.Errorf("cannot get pre-signed URL") - } - - rd, err := action.CreateArtifact(ctx) - - res, err := artifacts.PushBlob_(ctx, rd, create.SignedUploadUrl, 1024*1024*1, runtime.NumCPU()) - - finish, err := github.FinalizeArtifact(ctx, artifacts.FinalizeArtifactRequest{ - RunID: run, - JobRunID: job, - Name: action.name, - Size: res.Size, - Hash: res.SHA256Sum, - }) - if err != nil { - return fmt.Errorf("cannot finish artifact %w", err) - } - if !create.Ok { - return fmt.Errorf("cannot finish artifact upload") - } - - action.Noticef("created artifact: %s", finish.ArtifactId) - - return nil -} - -func (action *SaveArtifactAction) CreateArtifact(ctx context.Context) (io.Reader, error) { - rd, wr, err := os.Pipe() - go func() { - defer wr.Close() - - t := tar.NewWriter(wr) - t.AddFS(iox.NewGlobFS(os.DirFS(""), "")) - }() - return rd, err -} diff --git a/storage.go b/storage.go index f6a83ba..0712b7e 100644 --- a/storage.go +++ b/storage.go @@ -22,13 +22,12 @@ type PushResult struct { Size int64 } -func PushBlob(ctx context.Context, url string, data io.Reader, chunkSize, maxParallel int) (int, string, error) { - var empty string +func PushBlob(ctx context.Context, r io.Reader, blobURL string, chunkSize int64, parallelism int) (*PushResult, error) { if chunkSize <= 0 { chunkSize = 4 * 1024 * 1024 // 4MB default } - if maxParallel <= 0 { - maxParallel = 4 + if parallelism <= 0 { + parallelism = 4 } type chunk struct { @@ -36,19 +35,21 @@ func PushBlob(ctx context.Context, url string, data io.Reader, chunkSize, maxPar data []byte } wrkinp := make(chan chunk) - wrkerr := make(chan error, maxParallel) - blockIDs := make([]string, 0) + wrkerr := make(chan error, parallelism) + var blockMu sync.Mutex + blockIDs := make([]string, 0) + var wg sync.WaitGroup // Worker pool for parallel uploading - for i := 0; i < maxParallel; i++ { + for i := 0; i < parallelism; i++ { wg.Add(1) go func() { defer wg.Done() for c := range wrkinp { id := base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%06d", c.index))) - address := fmt.Sprintf("%s&comp=block&blockid=%s", url, id) + address := fmt.Sprintf("%s&comp=block&blockid=%s", blobURL, id) req, err := http.NewRequestWithContext(ctx, "PUT", address, bytes.NewReader(c.data)) if err != nil { wrkerr <- err @@ -77,11 +78,13 @@ func PushBlob(ctx context.Context, url string, data io.Reader, chunkSize, maxPar // Chunk the data and send to workers index := 0 hash := sha256.New() + var size int64 for { buf := make([]byte, chunkSize) - n, err := io.ReadFull(data, buf) + n, err := io.ReadFull(r, buf) if err == io.EOF || err == io.ErrUnexpectedEOF { if n > 0 { + size += int64(n) hash.Write(buf[:n]) wrkinp <- chunk{index: index, data: buf[:n]} index++ @@ -91,7 +94,7 @@ func PushBlob(ctx context.Context, url string, data io.Reader, chunkSize, maxPar if err != nil { close(wrkinp) wg.Wait() - return empty, err + return nil, err } hash.Write(buf) wrkinp <- chunk{index: index, data: buf} @@ -102,53 +105,51 @@ func PushBlob(ctx context.Context, url string, data io.Reader, chunkSize, maxPar select { case err := <-wrkerr: - return empty, err + return nil, err default: } - type BlockList struct { - Latest []string `xml:"Latest"` - } - type Latest struct { Value string `xml:",chardata"` } + type BlockList struct { + Latest []Latest `xml:"Latest"` + } // Sort block IDs to ensure correct order blockMu.Lock() defer blockMu.Unlock() if len(blockIDs) == 0 { - return empty, fmt.Errorf("no blocks uploaded") + return nil, fmt.Errorf("no blocks uploaded") } blockList := &BlockList{} for i := 0; i < len(blockIDs); i++ { - blockList.Latest = append(blockList.Blocks, xmlBlock{XMLName: xml.Name{Local: "Latest"}, ID: blockIDs[i]}) + blockList.Latest = append(blockList.Latest, Latest{Value: blockIDs[i]}) } blockListData, err := xml.Marshal(blockList) if err != nil { - return empty, err + return nil, err } - req, err := http.NewRequestWithContext(ctx, "PUT", fmt.Sprintf("%s&comp=blocklist", url), bytes.NewReader(blockListData)) + req, err := http.NewRequestWithContext(ctx, "PUT", fmt.Sprintf("%s&comp=blocklist", blobURL), bytes.NewReader(blockListData)) if err != nil { - return empty, err + return nil, err } req.Header.Set("Content-Type", "application/xml") req.ContentLength = int64(len(blockListData)) resp, err := http.DefaultClient.Do(req) if err != nil { - return empty, err + return nil, err } defer resp.Body.Close() + if resp.StatusCode != http.StatusCreated { body, _ := io.ReadAll(resp.Body) - return empty, fmt.Errorf("blocklist commit failed: %s body: %s", resp.Status, string(body)) + return nil, fmt.Errorf("blocklist commit failed: %s body: %s", resp.Status, string(body)) } - return hex.EncodeToString(hash.Sum(nil)), nil + return &PushResult{SHA256Sum: hex.EncodeToString(hash.Sum(nil)), Size: size}, nil } -func PushBlob_(ctx context.Context, r io.Reader, blobURL string, chunkSize int64, parallelism int) (*PushResult, error) { -} - -func PullBlob_(ctx context.Context, w io.Writer, blobURL string, chunkSize int64, parallelism int) (*PullResult, error) { +func PullBlob(ctx context.Context, w io.Writer, blobURL string, chunkSize int64, parallelism int) (*PullResult, error) { + return nil, nil }