ziggi/mstore
1
0
Fork 0
Code Pull Requests Actions Packages Projects Releases Wiki Activity

working commit

Browse Source
This commit is contained in:
Олег Бородин
2026-02-21 13:16:30 +02:00
parent cd37a4508c
commit d650d58a6d
1149 changed files with 116 additions and 722633 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vendor/github.com/google/go-containerregistry/pkg/v1/stream/README.md
Generated Vendored
-68
View File
@@ -1,68 +0,0 @@
# `stream`
[![GoDoc](https://godoc.org/github.com/google/go-containerregistry/pkg/v1/stream?status.svg)](https://godoc.org/github.com/google/go-containerregistry/pkg/v1/stream)
The `stream` package contains an implementation of
[`v1.Layer`](https://godoc.org/github.com/google/go-containerregistry/pkg/v1#Layer)
that supports _streaming_ access, i.e. the layer contents are read once and not
buffered.
## Usage
```go
package main
import (
"os"
"github.com/google/go-containerregistry/pkg/name"
"github.com/google/go-containerregistry/pkg/v1/remote"
"github.com/google/go-containerregistry/pkg/v1/stream"
)
// upload the contents of stdin as a layer to a local registry
func main() {
repo, err := name.NewRepository("localhost:5000/stream")
if err != nil {
panic(err)
}
layer := stream.NewLayer(os.Stdin)
if err := remote.WriteLayer(repo, layer); err != nil {
panic(err)
}
}
```
## Structure
This implements the layer portion of an [image
upload](/pkg/v1/remote#anatomy-of-an-image-upload). We launch a goroutine that
is responsible for hashing the uncompressed contents to compute the `DiffID`,
gzipping them to produce the `Compressed` contents, and hashing/counting the
bytes to produce the `Digest`/`Size`. This goroutine writes to an
`io.PipeWriter`, which blocks until `Compressed` reads the gzipped contents from
the corresponding `io.PipeReader`.
<p align="center">
<img src="/images/stream.dot.svg" />
</p>
## Caveats
This assumes that you have an uncompressed layer (i.e. a tarball) and would like
to compress it. Calling `Uncompressed` is always an error. Likewise, other
methods are invalid until the contents of `Compressed` have been completely
consumed and `Close`d.
Using a `stream.Layer` will likely not work without careful consideration. For
example, in the `mutate` package, we defer computing the manifest and config
file until they are actually called. This allows you to `mutate.Append` a
streaming layer to an image without accidentally consuming it. Similarly, in
`remote.Write`, if calling `Digest` on a layer fails, we attempt to upload the
layer anyway, understanding that we may be dealing with a `stream.Layer` whose
contents need to be uploaded before we can upload the config file.
Given the [structure](#structure) of how this is implemented, forgetting to
`Close` a `stream.Layer` will leak a goroutine.
vendor/github.com/google/go-containerregistry/pkg/v1/stream/layer.go
Generated Vendored
-275
View File
@@ -1,275 +0,0 @@
// Copyright 2018 Google LLC All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Package stream implements a single-pass streaming v1.Layer.
package stream
import (
"bufio"
"compress/gzip"
"crypto"
"encoding/hex"
"errors"
"hash"
"io"
"os"
"sync"
v1 "github.com/google/go-containerregistry/pkg/v1"
"github.com/google/go-containerregistry/pkg/v1/types"
)
var (
// ErrNotComputed is returned when the requested value is not yet
// computed because the stream has not been consumed yet.
ErrNotComputed = errors.New("value not computed until stream is consumed")
// ErrConsumed is returned by Compressed when the underlying stream has
// already been consumed and closed.
ErrConsumed = errors.New("stream was already consumed")
)
// Layer is a streaming implementation of v1.Layer.
type Layer struct {
blob io.ReadCloser
consumed bool
compression int
mu sync.Mutex
digest, diffID *v1.Hash
size int64
mediaType types.MediaType
}
var _ v1.Layer = (*Layer)(nil)
// LayerOption applies options to layer
type LayerOption func(*Layer)
// WithCompressionLevel sets the gzip compression. See `gzip.NewWriterLevel` for possible values.
func WithCompressionLevel(level int) LayerOption {
return func(l *Layer) {
l.compression = level
}
}
// WithMediaType is a functional option for overriding the layer's media type.
func WithMediaType(mt types.MediaType) LayerOption {
return func(l *Layer) {
l.mediaType = mt
}
}
// NewLayer creates a Layer from an io.ReadCloser.
func NewLayer(rc io.ReadCloser, opts ...LayerOption) *Layer {
layer := &Layer{
blob: rc,
compression: gzip.BestSpeed,
// We use DockerLayer for now as uncompressed layers
// are unimplemented
mediaType: types.DockerLayer,
}
for _, opt := range opts {
opt(layer)
}
return layer
}
// Digest implements v1.Layer.
func (l *Layer) Digest() (v1.Hash, error) {
l.mu.Lock()
defer l.mu.Unlock()
if l.digest == nil {
return v1.Hash{}, ErrNotComputed
}
return *l.digest, nil
}
// DiffID implements v1.Layer.
func (l *Layer) DiffID() (v1.Hash, error) {
l.mu.Lock()
defer l.mu.Unlock()
if l.diffID == nil {
return v1.Hash{}, ErrNotComputed
}
return *l.diffID, nil
}
// Size implements v1.Layer.
func (l *Layer) Size() (int64, error) {
l.mu.Lock()
defer l.mu.Unlock()
if l.size == 0 {
return 0, ErrNotComputed
}
return l.size, nil
}
// MediaType implements v1.Layer
func (l *Layer) MediaType() (types.MediaType, error) {
return l.mediaType, nil
}
// Uncompressed implements v1.Layer.
func (l *Layer) Uncompressed() (io.ReadCloser, error) {
return nil, errors.New("NYI: stream.Layer.Uncompressed is not implemented")
}
// Compressed implements v1.Layer.
func (l *Layer) Compressed() (io.ReadCloser, error) {
l.mu.Lock()
defer l.mu.Unlock()
if l.consumed {
return nil, ErrConsumed
}
return newCompressedReader(l)
}
// finalize sets the layer to consumed and computes all hash and size values.
func (l *Layer) finalize(uncompressed, compressed hash.Hash, size int64) error {
l.mu.Lock()
defer l.mu.Unlock()
diffID, err := v1.NewHash("sha256:" + hex.EncodeToString(uncompressed.Sum(nil)))
if err != nil {
return err
}
l.diffID = &diffID
digest, err := v1.NewHash("sha256:" + hex.EncodeToString(compressed.Sum(nil)))
if err != nil {
return err
}
l.digest = &digest
l.size = size
l.consumed = true
return nil
}
type compressedReader struct {
pr io.Reader
closer func() error
}
func newCompressedReader(l *Layer) (*compressedReader, error) {
// Collect digests of compressed and uncompressed stream and size of
// compressed stream.
h := crypto.SHA256.New()
zh := crypto.SHA256.New()
count := &countWriter{}
// gzip.Writer writes to the output stream via pipe, a hasher to
// capture compressed digest, and a countWriter to capture compressed
// size.
pr, pw := io.Pipe()
// Write compressed bytes to be read by the pipe.Reader, hashed by zh, and counted by count.
mw := io.MultiWriter(pw, zh, count)
// Buffer the output of the gzip writer so we don't have to wait on pr to keep writing.
// 64K ought to be small enough for anybody.
bw := bufio.NewWriterSize(mw, 2<<16)
zw, err := gzip.NewWriterLevel(bw, l.compression)
if err != nil {
return nil, err
}
doneDigesting := make(chan struct{})
cr := &compressedReader{
pr: pr,
closer: func() error {
// Immediately close pw without error. There are three ways to get
// here.
//
// 1. There was a copy error due from the underlying reader, in which
// case the error will not be overwritten.
// 2. Copying from the underlying reader completed successfully.
// 3. Close has been called before the underlying reader has been
// fully consumed. In this case pw must be closed in order to
// keep the flush of bw from blocking indefinitely.
//
// NOTE: pw.Close never returns an error. The signature is only to
// implement io.Closer.
_ = pw.Close()
// Close the inner ReadCloser.
//
// NOTE: net/http will call close on success, so if we've already
// closed the inner rc, it's not an error.
if err := l.blob.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
return err
}
// Finalize layer with its digest and size values.
<-doneDigesting
return l.finalize(h, zh, count.n)
},
}
go func() {
// Copy blob into the gzip writer, which also hashes and counts the
// size of the compressed output, and hasher of the raw contents.
_, copyErr := io.Copy(io.MultiWriter(h, zw), l.blob)
// Close the gzip writer once copying is done. If this is done in the
// Close method of compressedReader instead, then it can cause a panic
// when the compressedReader is closed before the blob is fully
// consumed and io.Copy in this goroutine is still blocking.
closeErr := zw.Close()
// Check errors from writing and closing streams.
if copyErr != nil {
close(doneDigesting)
pw.CloseWithError(copyErr)
return
}
if closeErr != nil {
close(doneDigesting)
pw.CloseWithError(closeErr)
return
}
// Flush the buffer once all writes are complete to the gzip writer.
if err := bw.Flush(); err != nil {
close(doneDigesting)
pw.CloseWithError(err)
return
}
// Notify closer that digests are done being written.
close(doneDigesting)
// Close the compressed reader to calculate digest/diffID/size. This
// will cause pr to return EOF which will cause readers of the
// Compressed stream to finish reading.
pw.CloseWithError(cr.Close())
}()
return cr, nil
}
func (cr *compressedReader) Read(b []byte) (int, error) { return cr.pr.Read(b) }
func (cr *compressedReader) Close() error { return cr.closer() }
// countWriter counts bytes written to it.
type countWriter struct{ n int64 }
func (c *countWriter) Write(p []byte) (int, error) {
c.n += int64(len(p))
return len(p), nil
}