util/imagetools: remove unused Resolver.ImageConfig

Signed-off-by: Sebastiaan van Stijn <github@gone.nl>
This commit is contained in:
Sebastiaan van Stijn 2023-07-17 12:33:54 +02:00
parent 089036da29
commit b9e25e82cf
No known key found for this signature in database
GPG Key ID: 76698F39D527CE8C
41 changed files with 0 additions and 4321 deletions

1
go.mod
View File

@ -76,7 +76,6 @@ require (
github.com/cenkalti/backoff/v4 v4.2.0 // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/cloudflare/cfssl v0.0.0-20181213083726-b94e044bb51e // indirect
github.com/containerd/ttrpc v1.2.2 // indirect
github.com/cyphar/filepath-securejoin v0.2.3 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/distribution/distribution/v3 v3.0.0-20230214150026-36d8c594d7aa // indirect

7
go.sum
View File

@ -136,7 +136,6 @@ github.com/containerd/nydus-snapshotter v0.8.2 h1:7SOrMU2YmLzfbsr5J7liMZJlNi5WT6
github.com/containerd/stargz-snapshotter v0.14.3 h1:OTUVZoPSPs8mGgmQUE1dqw3WX/3nrsmsurW7UPLWl1U=
github.com/containerd/stargz-snapshotter/estargz v0.14.3 h1:OqlDCK3ZVUO6C3B/5FSkDwbkEETK84kQgEeFwDC+62k=
github.com/containerd/ttrpc v1.2.2 h1:9vqZr0pxwOF5koz6N0N3kJ0zDHokrcPxIR/ZR2YFtOs=
github.com/containerd/ttrpc v1.2.2/go.mod h1:sIT6l32Ph/H9cvnJsfXM5drIVzTr5A2flTf1G5tYZak=
github.com/containerd/typeurl/v2 v2.1.1 h1:3Q4Pt7i8nYwy2KmQWIw2+1hTvwTE/6w9FqcttATPO/4=
github.com/containerd/typeurl/v2 v2.1.1/go.mod h1:IDp2JFvbwZ31H8dQbEIY7sDl2L3o3HZj1hsSQlywkQ0=
github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
@ -270,7 +269,6 @@ github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
@ -438,7 +436,6 @@ github.com/prometheus/common v0.42.0/go.mod h1:xBwqVerjNdUDjgODMpudtOMwlOwf2SaTr
github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
github.com/prometheus/procfs v0.0.3/go.mod h1:4A/X28fw3Fc593LaREMrKMqOKvUAntwMDaekg4FpcdQ=
github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
github.com/prometheus/procfs v0.9.0 h1:wzCHvIvM5SxWqYvwgVL7yJY8Lz3PKn49KQtpgMYJfhI=
github.com/prometheus/procfs v0.9.0/go.mod h1:+pB4zwohETzFnmlpe6yd2lSc+0/46IYZRB/chUwxUZY=
github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
@ -456,7 +453,6 @@ github.com/shibumi/go-pathspec v1.3.0 h1:QUyMZhFo0Md5B8zV8x2tesohbb5kfbpTi9rBnKh
github.com/shibumi/go-pathspec v1.3.0/go.mod h1:Xutfslp817l2I1cZvgcfeMQJG5QnU2lh5tVaaMCl3jE=
github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0=
github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=
github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
@ -636,7 +632,6 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.2.0 h1:PUR+T4wwASmuSTYdKjYHI5TD22Wy5ogLU5qZCOLxBrI=
golang.org/x/sync v0.2.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@ -654,7 +649,6 @@ golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20190726091711-fc99dfbffb4e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191001151750-bb3f8db39f24/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191204072324-ce4227a45e2e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191228213918-04cbcbbfeed8/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200113162924-86b910548bc1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@ -679,7 +673,6 @@ golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBc
golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

View File

@ -15,9 +15,7 @@ import (
clitypes "github.com/docker/cli/cli/config/types"
"github.com/docker/distribution/reference"
"github.com/moby/buildkit/util/contentutil"
"github.com/moby/buildkit/util/imageutil"
"github.com/moby/buildkit/util/tracing"
"github.com/opencontainers/go-digest"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/sirupsen/logrus"
)
@ -162,11 +160,3 @@ func RegistryAuthForRef(ref string, a Auth) (string, error) {
}
return base64.URLEncoding.EncodeToString(buf), nil
}
func (r *Resolver) ImageConfig(ctx context.Context, in string, platform *ocispec.Platform) (digest.Digest, []byte, error) {
in, _, err := r.Resolve(ctx, in)
if err != nil {
return "", nil, err
}
return imageutil.Config(ctx, in, r.resolver(), r.buffer, nil, platform)
}

View File

@ -1,73 +0,0 @@
/*
Copyright The containerd Authors.
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 identifiers provides common validation for identifiers and keys
// across containerd.
//
// Identifiers in containerd must be a alphanumeric, allowing limited
// underscores, dashes and dots.
//
// While the character set may be expanded in the future, identifiers
// are guaranteed to be safely used as filesystem path components.
package identifiers
import (
"fmt"
"regexp"
"github.com/containerd/containerd/errdefs"
)
const (
maxLength = 76
alphanum = `[A-Za-z0-9]+`
separators = `[._-]`
)
var (
// identifierRe defines the pattern for valid identifiers.
identifierRe = regexp.MustCompile(reAnchor(alphanum + reGroup(separators+reGroup(alphanum)) + "*"))
)
// Validate returns nil if the string s is a valid identifier.
//
// identifiers are similar to the domain name rules according to RFC 1035, section 2.3.1. However
// rules in this package are relaxed to allow numerals to follow period (".") and mixed case is
// allowed.
//
// In general identifiers that pass this validation should be safe for use as filesystem path components.
func Validate(s string) error {
if len(s) == 0 {
return fmt.Errorf("identifier must not be empty: %w", errdefs.ErrInvalidArgument)
}
if len(s) > maxLength {
return fmt.Errorf("identifier %q greater than maximum length (%d characters): %w", s, maxLength, errdefs.ErrInvalidArgument)
}
if !identifierRe.MatchString(s) {
return fmt.Errorf("identifier %q must match %v: %w", s, identifierRe, errdefs.ErrInvalidArgument)
}
return nil
}
func reGroup(s string) string {
return `(?:` + s + `)`
}
func reAnchor(s string) string {
return `^` + s + `$`
}

View File

@ -1,40 +0,0 @@
/*
Copyright The containerd Authors.
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 leases
import "context"
type leaseKey struct{}
// WithLease sets a given lease on the context
func WithLease(ctx context.Context, lid string) context.Context {
ctx = context.WithValue(ctx, leaseKey{}, lid)
// also store on the grpc headers so it gets picked up by any clients that
// are using this.
return withGRPCLeaseHeader(ctx, lid)
}
// FromContext returns the lease from the context.
func FromContext(ctx context.Context) (string, bool) {
lid, ok := ctx.Value(leaseKey{}).(string)
if !ok {
return fromGRPCHeader(ctx)
}
return lid, ok
}

View File

@ -1,58 +0,0 @@
/*
Copyright The containerd Authors.
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 leases
import (
"context"
"google.golang.org/grpc/metadata"
)
const (
// GRPCHeader defines the header name for specifying a containerd lease.
GRPCHeader = "containerd-lease"
)
func withGRPCLeaseHeader(ctx context.Context, lid string) context.Context {
// also store on the grpc headers so it gets picked up by any clients
// that are using this.
txheader := metadata.Pairs(GRPCHeader, lid)
md, ok := metadata.FromOutgoingContext(ctx) // merge with outgoing context.
if !ok {
md = txheader
} else {
// order ensures the latest is first in this list.
md = metadata.Join(txheader, md)
}
return metadata.NewOutgoingContext(ctx, md)
}
func fromGRPCHeader(ctx context.Context) (string, bool) {
// try to extract for use in grpc servers.
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return "", false
}
values := md[GRPCHeader]
if len(values) == 0 {
return "", false
}
return values[0], true
}

View File

@ -1,43 +0,0 @@
/*
Copyright The containerd Authors.
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 leases
import (
"crypto/rand"
"encoding/base64"
"fmt"
"time"
)
// WithRandomID sets the lease ID to a random unique value
func WithRandomID() Opt {
return func(l *Lease) error {
t := time.Now()
var b [3]byte
rand.Read(b[:])
l.ID = fmt.Sprintf("%d-%s", t.Nanosecond(), base64.URLEncoding.EncodeToString(b[:]))
return nil
}
}
// WithID sets the ID for the lease
func WithID(id string) Opt {
return func(l *Lease) error {
l.ID = id
return nil
}
}

View File

@ -1,91 +0,0 @@
/*
Copyright The containerd Authors.
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 leases
import (
"context"
"time"
)
// Opt is used to set options on a lease
type Opt func(*Lease) error
// DeleteOpt allows configuring a delete operation
type DeleteOpt func(context.Context, *DeleteOptions) error
// Manager is used to create, list, and remove leases
type Manager interface {
Create(context.Context, ...Opt) (Lease, error)
Delete(context.Context, Lease, ...DeleteOpt) error
List(context.Context, ...string) ([]Lease, error)
AddResource(context.Context, Lease, Resource) error
DeleteResource(context.Context, Lease, Resource) error
ListResources(context.Context, Lease) ([]Resource, error)
}
// Lease retains resources to prevent cleanup before
// the resources can be fully referenced.
type Lease struct {
ID string
CreatedAt time.Time
Labels map[string]string
}
// Resource represents low level resource of image, like content, ingest and
// snapshotter.
type Resource struct {
ID string
Type string
}
// DeleteOptions provide options on image delete
type DeleteOptions struct {
Synchronous bool
}
// SynchronousDelete is used to indicate that a lease deletion and removal of
// any unreferenced resources should occur synchronously before returning the
// result.
func SynchronousDelete(ctx context.Context, o *DeleteOptions) error {
o.Synchronous = true
return nil
}
// WithLabels merges labels on a lease
func WithLabels(labels map[string]string) Opt {
return func(l *Lease) error {
if l.Labels == nil {
l.Labels = map[string]string{}
}
for k, v := range labels {
l.Labels[k] = v
}
return nil
}
}
// WithExpiration sets an expiration on the lease
func WithExpiration(d time.Duration) Opt {
return func(l *Lease) error {
if l.Labels == nil {
l.Labels = map[string]string{}
}
l.Labels["containerd.io/gc.expire"] = time.Now().Add(d).Format(time.RFC3339)
return nil
}
}

View File

@ -1,78 +0,0 @@
/*
Copyright The containerd Authors.
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 namespaces
import (
"context"
"fmt"
"os"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/identifiers"
)
const (
// NamespaceEnvVar is the environment variable key name
NamespaceEnvVar = "CONTAINERD_NAMESPACE"
// Default is the name of the default namespace
Default = "default"
)
type namespaceKey struct{}
// WithNamespace sets a given namespace on the context
func WithNamespace(ctx context.Context, namespace string) context.Context {
ctx = context.WithValue(ctx, namespaceKey{}, namespace) // set our key for namespace
// also store on the grpc and ttrpc headers so it gets picked up by any clients that
// are using this.
return withTTRPCNamespaceHeader(withGRPCNamespaceHeader(ctx, namespace), namespace)
}
// NamespaceFromEnv uses the namespace defined in CONTAINERD_NAMESPACE or
// default
func NamespaceFromEnv(ctx context.Context) context.Context {
namespace := os.Getenv(NamespaceEnvVar)
if namespace == "" {
namespace = Default
}
return WithNamespace(ctx, namespace)
}
// Namespace returns the namespace from the context.
//
// The namespace is not guaranteed to be valid.
func Namespace(ctx context.Context) (string, bool) {
namespace, ok := ctx.Value(namespaceKey{}).(string)
if !ok {
if namespace, ok = fromGRPCHeader(ctx); !ok {
return fromTTRPCHeader(ctx)
}
}
return namespace, ok
}
// NamespaceRequired returns the valid namespace from the context or an error.
func NamespaceRequired(ctx context.Context) (string, error) {
namespace, ok := Namespace(ctx)
if !ok || namespace == "" {
return "", fmt.Errorf("namespace is required: %w", errdefs.ErrFailedPrecondition)
}
if err := identifiers.Validate(namespace); err != nil {
return "", fmt.Errorf("namespace validation: %w", err)
}
return namespace, nil
}

View File

@ -1,61 +0,0 @@
/*
Copyright The containerd Authors.
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 namespaces
import (
"context"
"google.golang.org/grpc/metadata"
)
const (
// GRPCHeader defines the header name for specifying a containerd namespace.
GRPCHeader = "containerd-namespace"
)
// NOTE(stevvooe): We can stub this file out if we don't want a grpc dependency here.
func withGRPCNamespaceHeader(ctx context.Context, namespace string) context.Context {
// also store on the grpc headers so it gets picked up by any clients that
// are using this.
nsheader := metadata.Pairs(GRPCHeader, namespace)
md, ok := metadata.FromOutgoingContext(ctx) // merge with outgoing context.
if !ok {
md = nsheader
} else {
// order ensures the latest is first in this list.
md = metadata.Join(nsheader, md)
}
return metadata.NewOutgoingContext(ctx, md)
}
func fromGRPCHeader(ctx context.Context) (string, bool) {
// try to extract for use in grpc servers.
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
// TODO(stevvooe): Check outgoing context?
return "", false
}
values := md[GRPCHeader]
if len(values) == 0 {
return "", false
}
return values[0], true
}

View File

@ -1,44 +0,0 @@
/*
Copyright The containerd Authors.
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 namespaces
import "context"
// Store provides introspection about namespaces.
//
// Note that these are slightly different than other objects, which are record
// oriented. A namespace is really just a name and a set of labels. Objects
// that belong to a namespace are returned when the namespace is assigned to a
// given context.
type Store interface {
Create(ctx context.Context, namespace string, labels map[string]string) error
Labels(ctx context.Context, namespace string) (map[string]string, error)
SetLabel(ctx context.Context, namespace, key, value string) error
List(ctx context.Context) ([]string, error)
// Delete removes the namespace. The namespace must be empty to be deleted.
Delete(ctx context.Context, namespace string, opts ...DeleteOpts) error
}
// DeleteInfo specifies information for the deletion of a namespace
type DeleteInfo struct {
// Name of the namespace
Name string
}
// DeleteOpts allows the caller to set options for namespace deletion
type DeleteOpts func(context.Context, *DeleteInfo) error

View File

@ -1,51 +0,0 @@
/*
Copyright The containerd Authors.
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 namespaces
import (
"context"
"github.com/containerd/ttrpc"
)
const (
// TTRPCHeader defines the header name for specifying a containerd namespace
TTRPCHeader = "containerd-namespace-ttrpc"
)
func copyMetadata(src ttrpc.MD) ttrpc.MD {
md := ttrpc.MD{}
for k, v := range src {
md[k] = append(md[k], v...)
}
return md
}
func withTTRPCNamespaceHeader(ctx context.Context, namespace string) context.Context {
md, ok := ttrpc.GetMetadata(ctx)
if !ok {
md = ttrpc.MD{}
} else {
md = copyMetadata(md)
}
md.Set(TTRPCHeader, namespace)
return ttrpc.WithMetadata(ctx, md)
}
func fromTTRPCHeader(ctx context.Context) (string, bool) {
return ttrpc.GetMetadataValue(ctx, TTRPCHeader)
}

View File

@ -1 +0,0 @@
*.go text eol=lf

View File

@ -1,13 +0,0 @@
# Binaries for programs and plugins
/bin/
*.exe
*.dll
*.so
*.dylib
# Test binary, build with `go test -c`
*.test
# Output of the go coverage tool, specifically when used with LiteIDE
*.out
coverage.txt

View File

@ -1,52 +0,0 @@
linters:
enable:
- staticcheck
- unconvert
- gofmt
- goimports
- revive
- ineffassign
- vet
- unused
- misspell
disable:
- errcheck
linters-settings:
revive:
ignore-generated-headers: true
rules:
- name: blank-imports
- name: context-as-argument
- name: context-keys-type
- name: dot-imports
- name: error-return
- name: error-strings
- name: error-naming
- name: exported
- name: if-return
- name: increment-decrement
- name: var-naming
arguments: [["UID", "GID"], []]
- name: var-declaration
- name: package-comments
- name: range
- name: receiver-naming
- name: time-naming
- name: unexported-return
- name: indent-error-flow
- name: errorf
- name: empty-block
- name: superfluous-else
- name: unused-parameter
- name: unreachable-code
- name: redefines-builtin-id
issues:
include:
- EXC0002
run:
timeout: 8m
skip-dirs:
- example

View File

@ -1,201 +0,0 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
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.

View File

@ -1,180 +0,0 @@
# Copyright The containerd Authors.
# 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.
# Go command to use for build
GO ?= go
INSTALL ?= install
# Root directory of the project (absolute path).
ROOTDIR=$(dir $(abspath $(lastword $(MAKEFILE_LIST))))
WHALE = "🇩"
ONI = "👹"
# Project binaries.
COMMANDS=protoc-gen-go-ttrpc protoc-gen-gogottrpc
ifdef BUILDTAGS
GO_BUILDTAGS = ${BUILDTAGS}
endif
GO_BUILDTAGS ?=
GO_TAGS=$(if $(GO_BUILDTAGS),-tags "$(strip $(GO_BUILDTAGS))",)
# Project packages.
PACKAGES=$(shell $(GO) list ${GO_TAGS} ./... | grep -v /example)
TESTPACKAGES=$(shell $(GO) list ${GO_TAGS} ./... | grep -v /cmd | grep -v /integration | grep -v /example)
BINPACKAGES=$(addprefix ./cmd/,$(COMMANDS))
#Replaces ":" (*nix), ";" (windows) with newline for easy parsing
GOPATHS=$(shell echo ${GOPATH} | tr ":" "\n" | tr ";" "\n")
TESTFLAGS_RACE=
GO_BUILD_FLAGS=
# See Golang issue re: '-trimpath': https://github.com/golang/go/issues/13809
GO_GCFLAGS=$(shell \
set -- ${GOPATHS}; \
echo "-gcflags=-trimpath=$${1}/src"; \
)
BINARIES=$(addprefix bin/,$(COMMANDS))
# Flags passed to `go test`
TESTFLAGS ?= $(TESTFLAGS_RACE) $(EXTRA_TESTFLAGS)
TESTFLAGS_PARALLEL ?= 8
# Use this to replace `go test` with, for instance, `gotestsum`
GOTEST ?= $(GO) test
.PHONY: clean all AUTHORS build binaries test integration generate protos check-protos coverage ci check help install vendor install-protobuf install-protobuild
.DEFAULT: default
# Forcibly set the default goal to all, in case an include above brought in a rule definition.
.DEFAULT_GOAL := all
all: binaries
check: proto-fmt ## run all linters
@echo "$(WHALE) $@"
GOGC=75 golangci-lint run
ci: check binaries check-protos coverage # coverage-integration ## to be used by the CI
AUTHORS: .mailmap .git/HEAD
git log --format='%aN <%aE>' | sort -fu > $@
generate: protos
@echo "$(WHALE) $@"
@PATH="${ROOTDIR}/bin:${PATH}" $(GO) generate -x ${PACKAGES}
protos: bin/protoc-gen-gogottrpc bin/protoc-gen-go-ttrpc ## generate protobuf
@echo "$(WHALE) $@"
@(PATH="${ROOTDIR}/bin:${PATH}" protobuild --quiet ${PACKAGES})
check-protos: protos ## check if protobufs needs to be generated again
@echo "$(WHALE) $@"
@test -z "$$(git status --short | grep ".pb.go" | tee /dev/stderr)" || \
((git diff | cat) && \
(echo "$(ONI) please run 'make protos' when making changes to proto files" && false))
check-api-descriptors: protos ## check that protobuf changes aren't present.
@echo "$(WHALE) $@"
@test -z "$$(git status --short | grep ".pb.txt" | tee /dev/stderr)" || \
((git diff $$(find . -name '*.pb.txt') | cat) && \
(echo "$(ONI) please run 'make protos' when making changes to proto files and check-in the generated descriptor file changes" && false))
proto-fmt: ## check format of proto files
@echo "$(WHALE) $@"
@test -z "$$(find . -name '*.proto' -type f -exec grep -Hn -e "^ " {} \; | tee /dev/stderr)" || \
(echo "$(ONI) please indent proto files with tabs only" && false)
@test -z "$$(find . -name '*.proto' -type f -exec grep -Hn "Meta meta = " {} \; | grep -v '(gogoproto.nullable) = false' | tee /dev/stderr)" || \
(echo "$(ONI) meta fields in proto files must have option (gogoproto.nullable) = false" && false)
build: ## build the go packages
@echo "$(WHALE) $@"
@$(GO) build ${DEBUG_GO_GCFLAGS} ${GO_GCFLAGS} ${GO_BUILD_FLAGS} ${EXTRA_FLAGS} ${PACKAGES}
test: ## run tests, except integration tests and tests that require root
@echo "$(WHALE) $@"
@$(GOTEST) ${TESTFLAGS} ${TESTPACKAGES}
integration: ## run integration tests
@echo "$(WHALE) $@"
@cd "${ROOTDIR}/integration" && $(GOTEST) -v ${TESTFLAGS} -parallel ${TESTFLAGS_PARALLEL} .
benchmark: ## run benchmarks tests
@echo "$(WHALE) $@"
@$(GO) test ${TESTFLAGS} -bench . -run Benchmark
FORCE:
define BUILD_BINARY
@echo "$(WHALE) $@"
@$(GO) build ${DEBUG_GO_GCFLAGS} ${GO_GCFLAGS} ${GO_BUILD_FLAGS} -o $@ ${GO_TAGS} ./$<
endef
# Build a binary from a cmd.
bin/%: cmd/% FORCE
$(call BUILD_BINARY)
binaries: $(BINARIES) ## build binaries
@echo "$(WHALE) $@"
clean: ## clean up binaries
@echo "$(WHALE) $@"
@rm -f $(BINARIES)
install: ## install binaries
@echo "$(WHALE) $@ $(BINPACKAGES)"
@$(GO) install $(BINPACKAGES)
install-protobuf:
@echo "$(WHALE) $@"
@script/install-protobuf
install-protobuild:
@echo "$(WHALE) $@"
@$(GO) install google.golang.org/protobuf/cmd/protoc-gen-go@v1.28.1
@$(GO) install github.com/containerd/protobuild@14832ccc41429f5c4f81028e5af08aa233a219cf
coverage: ## generate coverprofiles from the unit tests, except tests that require root
@echo "$(WHALE) $@"
@rm -f coverage.txt
@$(GO) test ${TESTFLAGS} ${TESTPACKAGES} 2> /dev/null
@( for pkg in ${PACKAGES}; do \
$(GO) test ${TESTFLAGS} \
-cover \
-coverprofile=profile.out \
-covermode=atomic $$pkg || exit; \
if [ -f profile.out ]; then \
cat profile.out >> coverage.txt; \
rm profile.out; \
fi; \
done )
vendor: ## ensure all the go.mod/go.sum files are up-to-date
@echo "$(WHALE) $@"
@$(GO) mod tidy
@$(GO) mod verify
verify-vendor: ## verify if all the go.mod/go.sum files are up-to-date
@echo "$(WHALE) $@"
@$(GO) mod tidy
@$(GO) mod verify
@test -z "$$(git status --short | grep "go.sum" | tee /dev/stderr)" || \
((git diff | cat) && \
(echo "$(ONI) make sure to checkin changes after go mod tidy" && false))
help: ## this help
@awk 'BEGIN {FS = ":.*?## "} /^[a-zA-Z_-]+:.*?## / {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}' $(MAKEFILE_LIST) | sort

View File

@ -1,240 +0,0 @@
# Protocol Specification
The ttrpc protocol is client/server protocol to support multiple request streams
over a single connection with lightweight framing. The client represents the
process which initiated the underlying connection and the server is the process
which accepted the connection. The protocol is currently defined as
asymmetrical, with clients sending requests and servers sending responses. Both
clients and servers are able to send stream data. The roles are also used in
determining the stream identifiers, with client initiated streams using odd
number identifiers and server initiated using even number. The protocol may be
extended in the future to support server initiated streams, that is not
supported in the latest version.
## Purpose
The ttrpc protocol is designed to be lightweight and optimized for low latency
and reliable connections between processes on the same host. The protocol does
not include features for handling unreliable connections such as handshakes,
resets, pings, or flow control. The protocol is designed to make low-overhead
implementations as simple as possible. It is not intended as a suitable
replacement for HTTP2/3 over the network.
## Message Frame
Each Message Frame consists of a 10-byte message header followed
by message data. The data length and stream ID are both big-endian
4-byte unsigned integers. The message type is an unsigned 1-byte
integer. The flags are also an unsigned 1-byte integer and
use is defined by the message type.
+---------------------------------------------------------------+
| Data Length (32) |
+---------------------------------------------------------------+
| Stream ID (32) |
+---------------+-----------------------------------------------+
| Msg Type (8) |
+---------------+
| Flags (8) |
+---------------+-----------------------------------------------+
| Data (*) |
+---------------------------------------------------------------+
The Data Length field represents the number of bytes in the Data field. The
total frame size will always be Data Length + 10 bytes. The maximum data length
is 4MB and any larger size should be rejected. Due to the maximum data size
being less than 16MB, the first frame byte should always be zero. This first
byte should be considered reserved for future use.
The Stream ID must be odd for client initiated streams and even for server
initiated streams. Server initiated streams are not currently supported.
## Mesage Types
| Message Type | Name | Description |
|--------------|----------|----------------------------------|
| 0x01 | Request | Initiates stream |
| 0x02 | Response | Final stream data and terminates |
| 0x03 | Data | Stream data |
### Request
The request message is used to initiate stream and send along request data for
properly routing and handling the stream. The stream may indicate unary without
any inbound or outbound stream data with only a response is expected on the
stream. The request may also indicate the stream is still open for more data and
no response is expected until data is finished. If the remote indicates the
stream is closed, the request may be considered non-unary but without anymore
stream data sent. In the case of `remote closed`, the remote still expects to
receive a response or stream data. For compatibility with non streaming clients,
a request with empty flags indicates a unary request.
#### Request Flags
| Flag | Name | Description |
|------|-----------------|--------------------------------------------------|
| 0x01 | `remote closed` | Non-unary, but no more data expected from remote |
| 0x02 | `remote open` | Non-unary, remote is still sending data |
### Response
The response message is used to end a stream with data, an empty response, or
an error. A response message is the only expected message after a unary request.
A non-unary request does not require a response message if the server is sending
back stream data. A non-unary stream may return a single response message but no
other stream data may follow.
#### Response Flags
No response flags are defined at this time, flags should be empty.
### Data
The data message is used to send data on an already initialized stream. Either
client or server may send data. A data message is not allowed on a unary stream.
A data message should not be sent after indicating `remote closed` to the peer.
The last data message on a stream must set the `remote closed` flag.
The `no data` flag is used to indicate that the data message does not include
any data. This is normally used with the `remote closed` flag to indicate the
stream is now closed without transmitting any data. Since ttrpc normally
transmits a single object per message, a zero length data message may be
interpreted as an empty object. For example, transmitting the number zero as a
protobuf message ends up with a data length of zero, but the message is still
considered data and should be processed.
#### Data Flags
| Flag | Name | Description |
|------|-----------------|-----------------------------------|
| 0x01 | `remote closed` | No more data expected from remote |
| 0x04 | `no data` | This message does not have data |
## Streaming
All ttrpc requests use streams to transfer data. Unary streams will only have
two messages sent per stream, a request from a client and a response from the
server. Non-unary streams, however, may send any numbers of messages from the
client and the server. This makes stream management more complicated than unary
streams since both client and server need to track additional state. To keep
this management as simple as possible, ttrpc minimizes the number of states and
uses two flags instead of control frames. Each stream has two states while a
stream is still alive: `local closed` and `remote closed`. Each peer considers
local and remote from their own perspective and sets flags from the other peer's
perspective. For example, if a client sends a data frame with the
`remote closed` flag, that is indicating that the client is now `local closed`
and the server will be `remote closed`. A unary operation does not need to send
these flags since each received message always indicates `remote closed`. Once a
peer is both `local closed` and `remote closed`, the stream is considered
finished and may be cleaned up.
Due to the asymmetric nature of the current protocol, a client should
always be in the `local closed` state before `remote closed` and a server should
always be in the `remote closed` state before `local closed`. This happens
because the client is always initiating requests and a client always expects a
final response back from a server to indicate the initiated request has been
fulfilled. This may mean server sends a final empty response to finish a stream
even after it has already completed sending data before the client.
### Unary State Diagram
+--------+ +--------+
| Client | | Server |
+---+----+ +----+---+
| +---------+ |
local >---------------+ Request +--------------------> remote
closed | +---------+ | closed
| |
| +----------+ |
finished <--------------+ Response +--------------------< finished
| +----------+ |
| |
### Non-Unary State Diagrams
RC: `remote closed` flag
RO: `remote open` flag
+--------+ +--------+
| Client | | Server |
+---+----+ +----+---+
| +--------------+ |
>-------------+ Request [RO] +----------------->
| +--------------+ |
| |
| +------+ |
>-----------------+ Data +--------------------->
| +------+ |
| |
| +-----------+ |
local >---------------+ Data [RC] +------------------> remote
closed | +-----------+ | closed
| |
| +----------+ |
finished <--------------+ Response +--------------------< finished
| +----------+ |
| |
+--------+ +--------+
| Client | | Server |
+---+----+ +----+---+
| +--------------+ |
local >-------------+ Request [RC] +-----------------> remote
closed | +--------------+ | closed
| |
| +------+ |
<-----------------+ Data +---------------------<
| +------+ |
| |
| +-----------+ |
finished <---------------+ Data [RC] +------------------< finished
| +-----------+ |
| |
+--------+ +--------+
| Client | | Server |
+---+----+ +----+---+
| +--------------+ |
>-------------+ Request [RO] +----------------->
| +--------------+ |
| |
| +------+ |
>-----------------+ Data +--------------------->
| +------+ |
| |
| +------+ |
<-----------------+ Data +---------------------<
| +------+ |
| |
| +------+ |
>-----------------+ Data +--------------------->
| +------+ |
| |
| +-----------+ |
local >---------------+ Data [RC] +------------------> remote
closed | +-----------+ | closed
| |
| +------+ |
<-----------------+ Data +---------------------<
| +------+ |
| |
| +-----------+ |
finished <---------------+ Data [RC] +------------------< finished
| +-----------+ |
| |
## RPC
While this protocol is defined primarily to support Remote Procedure Calls, the
protocol does not define the request and response types beyond the messages
defined in the protocol. The implementation provides a default protobuf
definition of request and response which may be used for cross language rpc.
All implementations should at least define a request type which support
routing by procedure name and a response type which supports call status.
## Version History
| Version | Features |
|---------|---------------------|
| 1.0 | Unary requests only |
| 1.2 | Streaming support |

View File

@ -1,28 +0,0 @@
version = "2"
generators = ["go"]
# Control protoc include paths. Below are usually some good defaults, but feel
# free to try it without them if it works for your project.
[includes]
# Include paths that will be added before all others. Typically, you want to
# treat the root of the project as an include, but this may not be necessary.
before = ["."]
# Paths that will be added untouched to the end of the includes. We use
# `/usr/local/include` to pickup the common install location of protobuf.
# This is the default.
after = ["/usr/local/include"]
# This section maps protobuf imports to Go packages. These will become
# `-M` directives in the call to the go protobuf generator.
[packages]
"google/protobuf/any.proto" = "github.com/gogo/protobuf/types"
"proto/status.proto" = "google.golang.org/genproto/googleapis/rpc/status"
[[overrides]]
# enable ttrpc and disable fieldpath and grpc for the shim
prefixes = ["github.com/containerd/ttrpc/integration/streaming"]
generators = ["go", "go-ttrpc"]
[overrides.parameters.go-ttrpc]
prefix = "TTRPC"

View File

@ -1,59 +0,0 @@
# ttrpc
[![Build Status](https://github.com/containerd/ttrpc/workflows/CI/badge.svg)](https://github.com/containerd/ttrpc/actions?query=workflow%3ACI)
GRPC for low-memory environments.
The existing grpc-go project requires a lot of memory overhead for importing
packages and at runtime. While this is great for many services with low density
requirements, this can be a problem when running a large number of services on
a single machine or on a machine with a small amount of memory.
Using the same GRPC definitions, this project reduces the binary size and
protocol overhead required. We do this by eliding the `net/http`, `net/http2`
and `grpc` package used by grpc replacing it with a lightweight framing
protocol. The result are smaller binaries that use less resident memory with
the same ease of use as GRPC.
Please note that while this project supports generating either end of the
protocol, the generated service definitions will be incompatible with regular
GRPC services, as they do not speak the same protocol.
# Protocol
See the [protocol specification](./PROTOCOL.md).
# Usage
Create a gogo vanity binary (see
[`cmd/protoc-gen-gogottrpc/main.go`](cmd/protoc-gen-gogottrpc/main.go) for an
example with the ttrpc plugin enabled.
It's recommended to use [`protobuild`](https://github.com/containerd/protobuild)
to build the protobufs for this project, but this will work with protoc
directly, if required.
# Differences from GRPC
- The protocol stack has been replaced with a lighter protocol that doesn't
require http, http2 and tls.
- The client and server interface are identical whereas in GRPC there is a
client and server interface that are different.
- The Go stdlib context package is used instead.
# Status
TODO:
- [ ] Add testing under concurrent load to ensure
- [ ] Verify connection error handling
# Project details
ttrpc is a containerd sub-project, licensed under the [Apache 2.0 license](./LICENSE).
As a containerd sub-project, you will find the:
* [Project governance](https://github.com/containerd/project/blob/main/GOVERNANCE.md),
* [Maintainers](https://github.com/containerd/project/blob/main/MAINTAINERS),
* and [Contributing guidelines](https://github.com/containerd/project/blob/main/CONTRIBUTING.md)
information in our [`containerd/project`](https://github.com/containerd/project) repository.

View File

@ -1,182 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"bufio"
"encoding/binary"
"fmt"
"io"
"net"
"sync"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
const (
messageHeaderLength = 10
messageLengthMax = 4 << 20
)
type messageType uint8
const (
messageTypeRequest messageType = 0x1
messageTypeResponse messageType = 0x2
messageTypeData messageType = 0x3
)
func (mt messageType) String() string {
switch mt {
case messageTypeRequest:
return "request"
case messageTypeResponse:
return "response"
case messageTypeData:
return "data"
default:
return "unknown"
}
}
const (
flagRemoteClosed uint8 = 0x1
flagRemoteOpen uint8 = 0x2
flagNoData uint8 = 0x4
)
// messageHeader represents the fixed-length message header of 10 bytes sent
// with every request.
type messageHeader struct {
Length uint32 // length excluding this header. b[:4]
StreamID uint32 // identifies which request stream message is a part of. b[4:8]
Type messageType // message type b[8]
Flags uint8 // type specific flags b[9]
}
func readMessageHeader(p []byte, r io.Reader) (messageHeader, error) {
_, err := io.ReadFull(r, p[:messageHeaderLength])
if err != nil {
return messageHeader{}, err
}
return messageHeader{
Length: binary.BigEndian.Uint32(p[:4]),
StreamID: binary.BigEndian.Uint32(p[4:8]),
Type: messageType(p[8]),
Flags: p[9],
}, nil
}
func writeMessageHeader(w io.Writer, p []byte, mh messageHeader) error {
binary.BigEndian.PutUint32(p[:4], mh.Length)
binary.BigEndian.PutUint32(p[4:8], mh.StreamID)
p[8] = byte(mh.Type)
p[9] = mh.Flags
_, err := w.Write(p[:])
return err
}
var buffers sync.Pool
type channel struct {
conn net.Conn
bw *bufio.Writer
br *bufio.Reader
hrbuf [messageHeaderLength]byte // avoid alloc when reading header
hwbuf [messageHeaderLength]byte
}
func newChannel(conn net.Conn) *channel {
return &channel{
conn: conn,
bw: bufio.NewWriter(conn),
br: bufio.NewReader(conn),
}
}
// recv a message from the channel. The returned buffer contains the message.
//
// If a valid grpc status is returned, the message header
// returned will be valid and caller should send that along to
// the correct consumer. The bytes on the underlying channel
// will be discarded.
func (ch *channel) recv() (messageHeader, []byte, error) {
mh, err := readMessageHeader(ch.hrbuf[:], ch.br)
if err != nil {
return messageHeader{}, nil, err
}
if mh.Length > uint32(messageLengthMax) {
if _, err := ch.br.Discard(int(mh.Length)); err != nil {
return mh, nil, fmt.Errorf("failed to discard after receiving oversized message: %w", err)
}
return mh, nil, status.Errorf(codes.ResourceExhausted, "message length %v exceed maximum message size of %v", mh.Length, messageLengthMax)
}
var p []byte
if mh.Length > 0 {
p = ch.getmbuf(int(mh.Length))
if _, err := io.ReadFull(ch.br, p); err != nil {
return messageHeader{}, nil, fmt.Errorf("failed reading message: %w", err)
}
}
return mh, p, nil
}
func (ch *channel) send(streamID uint32, t messageType, flags uint8, p []byte) error {
// TODO: Error on send rather than on recv
//if len(p) > messageLengthMax {
// return status.Errorf(codes.InvalidArgument, "refusing to send, message length %v exceed maximum message size of %v", len(p), messageLengthMax)
//}
if err := writeMessageHeader(ch.bw, ch.hwbuf[:], messageHeader{Length: uint32(len(p)), StreamID: streamID, Type: t, Flags: flags}); err != nil {
return err
}
if len(p) > 0 {
_, err := ch.bw.Write(p)
if err != nil {
return err
}
}
return ch.bw.Flush()
}
func (ch *channel) getmbuf(size int) []byte {
// we can't use the standard New method on pool because we want to allocate
// based on size.
b, ok := buffers.Get().(*[]byte)
if !ok || cap(*b) < size {
// TODO(stevvooe): It may be better to allocate these in fixed length
// buckets to reduce fragmentation but its not clear that would help
// with performance. An ilogb approach or similar would work well.
bb := make([]byte, size)
b = &bb
} else {
*b = (*b)[:size]
}
return *b
}
func (ch *channel) putmbuf(p []byte) {
buffers.Put(&p)
}

View File

@ -1,512 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"context"
"errors"
"fmt"
"io"
"net"
"strings"
"sync"
"syscall"
"time"
"github.com/sirupsen/logrus"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
)
// Client for a ttrpc server
type Client struct {
codec codec
conn net.Conn
channel *channel
streamLock sync.RWMutex
streams map[streamID]*stream
nextStreamID streamID
sendLock sync.Mutex
ctx context.Context
closed func()
closeOnce sync.Once
userCloseFunc func()
userCloseWaitCh chan struct{}
interceptor UnaryClientInterceptor
}
// ClientOpts configures a client
type ClientOpts func(c *Client)
// WithOnClose sets the close func whenever the client's Close() method is called
func WithOnClose(onClose func()) ClientOpts {
return func(c *Client) {
c.userCloseFunc = onClose
}
}
// WithUnaryClientInterceptor sets the provided client interceptor
func WithUnaryClientInterceptor(i UnaryClientInterceptor) ClientOpts {
return func(c *Client) {
c.interceptor = i
}
}
// NewClient creates a new ttrpc client using the given connection
func NewClient(conn net.Conn, opts ...ClientOpts) *Client {
ctx, cancel := context.WithCancel(context.Background())
channel := newChannel(conn)
c := &Client{
codec: codec{},
conn: conn,
channel: channel,
streams: make(map[streamID]*stream),
nextStreamID: 1,
closed: cancel,
ctx: ctx,
userCloseFunc: func() {},
userCloseWaitCh: make(chan struct{}),
interceptor: defaultClientInterceptor,
}
for _, o := range opts {
o(c)
}
go c.run()
return c
}
func (c *Client) send(sid uint32, mt messageType, flags uint8, b []byte) error {
c.sendLock.Lock()
defer c.sendLock.Unlock()
return c.channel.send(sid, mt, flags, b)
}
// Call makes a unary request and returns with response
func (c *Client) Call(ctx context.Context, service, method string, req, resp interface{}) error {
payload, err := c.codec.Marshal(req)
if err != nil {
return err
}
var (
creq = &Request{
Service: service,
Method: method,
Payload: payload,
// TODO: metadata from context
}
cresp = &Response{}
)
if metadata, ok := GetMetadata(ctx); ok {
metadata.setRequest(creq)
}
if dl, ok := ctx.Deadline(); ok {
creq.TimeoutNano = time.Until(dl).Nanoseconds()
}
info := &UnaryClientInfo{
FullMethod: fullPath(service, method),
}
if err := c.interceptor(ctx, creq, cresp, info, c.dispatch); err != nil {
return err
}
if err := c.codec.Unmarshal(cresp.Payload, resp); err != nil {
return err
}
if cresp.Status != nil && cresp.Status.Code != int32(codes.OK) {
return status.ErrorProto(cresp.Status)
}
return nil
}
// StreamDesc describes the stream properties, whether the stream has
// a streaming client, a streaming server, or both
type StreamDesc struct {
StreamingClient bool
StreamingServer bool
}
// ClientStream is used to send or recv messages on the underlying stream
type ClientStream interface {
CloseSend() error
SendMsg(m interface{}) error
RecvMsg(m interface{}) error
}
type clientStream struct {
ctx context.Context
s *stream
c *Client
desc *StreamDesc
localClosed bool
remoteClosed bool
}
func (cs *clientStream) CloseSend() error {
if !cs.desc.StreamingClient {
return fmt.Errorf("%w: cannot close non-streaming client", ErrProtocol)
}
if cs.localClosed {
return ErrStreamClosed
}
err := cs.s.send(messageTypeData, flagRemoteClosed|flagNoData, nil)
if err != nil {
return filterCloseErr(err)
}
cs.localClosed = true
return nil
}
func (cs *clientStream) SendMsg(m interface{}) error {
if !cs.desc.StreamingClient {
return fmt.Errorf("%w: cannot send data from non-streaming client", ErrProtocol)
}
if cs.localClosed {
return ErrStreamClosed
}
var (
payload []byte
err error
)
if m != nil {
payload, err = cs.c.codec.Marshal(m)
if err != nil {
return err
}
}
err = cs.s.send(messageTypeData, 0, payload)
if err != nil {
return filterCloseErr(err)
}
return nil
}
func (cs *clientStream) RecvMsg(m interface{}) error {
if cs.remoteClosed {
return io.EOF
}
var msg *streamMessage
select {
case <-cs.ctx.Done():
return cs.ctx.Err()
case <-cs.s.recvClose:
// If recv has a pending message, process that first
select {
case msg = <-cs.s.recv:
default:
return cs.s.recvErr
}
case msg = <-cs.s.recv:
}
if msg.header.Type == messageTypeResponse {
resp := &Response{}
err := proto.Unmarshal(msg.payload[:msg.header.Length], resp)
// return the payload buffer for reuse
cs.c.channel.putmbuf(msg.payload)
if err != nil {
return err
}
if err := cs.c.codec.Unmarshal(resp.Payload, m); err != nil {
return err
}
if resp.Status != nil && resp.Status.Code != int32(codes.OK) {
return status.ErrorProto(resp.Status)
}
cs.c.deleteStream(cs.s)
cs.remoteClosed = true
return nil
} else if msg.header.Type == messageTypeData {
if !cs.desc.StreamingServer {
cs.c.deleteStream(cs.s)
cs.remoteClosed = true
return fmt.Errorf("received data from non-streaming server: %w", ErrProtocol)
}
if msg.header.Flags&flagRemoteClosed == flagRemoteClosed {
cs.c.deleteStream(cs.s)
cs.remoteClosed = true
if msg.header.Flags&flagNoData == flagNoData {
return io.EOF
}
}
err := cs.c.codec.Unmarshal(msg.payload[:msg.header.Length], m)
cs.c.channel.putmbuf(msg.payload)
if err != nil {
return err
}
return nil
}
return fmt.Errorf("unexpected %q message received: %w", msg.header.Type, ErrProtocol)
}
// Close closes the ttrpc connection and underlying connection
func (c *Client) Close() error {
c.closeOnce.Do(func() {
c.closed()
c.conn.Close()
})
return nil
}
// UserOnCloseWait is used to blocks untils the user's on-close callback
// finishes.
func (c *Client) UserOnCloseWait(ctx context.Context) error {
select {
case <-c.userCloseWaitCh:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (c *Client) run() {
err := c.receiveLoop()
c.Close()
c.cleanupStreams(err)
c.userCloseFunc()
close(c.userCloseWaitCh)
}
func (c *Client) receiveLoop() error {
for {
select {
case <-c.ctx.Done():
return ErrClosed
default:
var (
msg = &streamMessage{}
err error
)
msg.header, msg.payload, err = c.channel.recv()
if err != nil {
_, ok := status.FromError(err)
if !ok {
// treat all errors that are not an rpc status as terminal.
// all others poison the connection.
return filterCloseErr(err)
}
}
sid := streamID(msg.header.StreamID)
s := c.getStream(sid)
if s == nil {
logrus.WithField("stream", sid).Errorf("ttrpc: received message on inactive stream")
continue
}
if err != nil {
s.closeWithError(err)
} else {
if err := s.receive(c.ctx, msg); err != nil {
logrus.WithError(err).WithField("stream", sid).Errorf("ttrpc: failed to handle message")
}
}
}
}
}
// createStream creates a new stream and registers it with the client
// Introduce stream types for multiple or single response
func (c *Client) createStream(flags uint8, b []byte) (*stream, error) {
c.streamLock.Lock()
// Check if closed since lock acquired to prevent adding
// anything after cleanup completes
select {
case <-c.ctx.Done():
c.streamLock.Unlock()
return nil, ErrClosed
default:
}
// Stream ID should be allocated at same time
s := newStream(c.nextStreamID, c)
c.streams[s.id] = s
c.nextStreamID = c.nextStreamID + 2
c.sendLock.Lock()
defer c.sendLock.Unlock()
c.streamLock.Unlock()
if err := c.channel.send(uint32(s.id), messageTypeRequest, flags, b); err != nil {
return s, filterCloseErr(err)
}
return s, nil
}
func (c *Client) deleteStream(s *stream) {
c.streamLock.Lock()
delete(c.streams, s.id)
c.streamLock.Unlock()
s.closeWithError(nil)
}
func (c *Client) getStream(sid streamID) *stream {
c.streamLock.RLock()
s := c.streams[sid]
c.streamLock.RUnlock()
return s
}
func (c *Client) cleanupStreams(err error) {
c.streamLock.Lock()
defer c.streamLock.Unlock()
for sid, s := range c.streams {
s.closeWithError(err)
delete(c.streams, sid)
}
}
// filterCloseErr rewrites EOF and EPIPE errors to ErrClosed. Use when
// returning from call or handling errors from main read loop.
//
// This purposely ignores errors with a wrapped cause.
func filterCloseErr(err error) error {
switch {
case err == nil:
return nil
case err == io.EOF:
return ErrClosed
case errors.Is(err, io.ErrClosedPipe):
return ErrClosed
case errors.Is(err, io.EOF):
return ErrClosed
case strings.Contains(err.Error(), "use of closed network connection"):
return ErrClosed
default:
// if we have an epipe on a write or econnreset on a read , we cast to errclosed
var oerr *net.OpError
if errors.As(err, &oerr) {
if (oerr.Op == "write" && errors.Is(err, syscall.EPIPE)) ||
(oerr.Op == "read" && errors.Is(err, syscall.ECONNRESET)) {
return ErrClosed
}
}
}
return err
}
// NewStream creates a new stream with the given stream descriptor to the
// specified service and method. If not a streaming client, the request object
// may be provided.
func (c *Client) NewStream(ctx context.Context, desc *StreamDesc, service, method string, req interface{}) (ClientStream, error) {
var payload []byte
if req != nil {
var err error
payload, err = c.codec.Marshal(req)
if err != nil {
return nil, err
}
}
request := &Request{
Service: service,
Method: method,
Payload: payload,
// TODO: metadata from context
}
p, err := c.codec.Marshal(request)
if err != nil {
return nil, err
}
var flags uint8
if desc.StreamingClient {
flags = flagRemoteOpen
} else {
flags = flagRemoteClosed
}
s, err := c.createStream(flags, p)
if err != nil {
return nil, err
}
return &clientStream{
ctx: ctx,
s: s,
c: c,
desc: desc,
}, nil
}
func (c *Client) dispatch(ctx context.Context, req *Request, resp *Response) error {
p, err := c.codec.Marshal(req)
if err != nil {
return err
}
s, err := c.createStream(0, p)
if err != nil {
return err
}
defer c.deleteStream(s)
var msg *streamMessage
select {
case <-ctx.Done():
return ctx.Err()
case <-c.ctx.Done():
return ErrClosed
case <-s.recvClose:
// If recv has a pending message, process that first
select {
case msg = <-s.recv:
default:
return s.recvErr
}
case msg = <-s.recv:
}
if msg.header.Type == messageTypeResponse {
err = proto.Unmarshal(msg.payload[:msg.header.Length], resp)
} else {
err = fmt.Errorf("unexpected %q message received: %w", msg.header.Type, ErrProtocol)
}
// return the payload buffer for reuse
c.channel.putmbuf(msg.payload)
return err
}

View File

@ -1,43 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"fmt"
"google.golang.org/protobuf/proto"
)
type codec struct{}
func (c codec) Marshal(msg interface{}) ([]byte, error) {
switch v := msg.(type) {
case proto.Message:
return proto.Marshal(v)
default:
return nil, fmt.Errorf("ttrpc: cannot marshal unknown type: %T", msg)
}
}
func (c codec) Unmarshal(p []byte, msg interface{}) error {
switch v := msg.(type) {
case proto.Message:
return proto.Unmarshal(p, v)
default:
return fmt.Errorf("ttrpc: cannot unmarshal into unknown type: %T", msg)
}
}

View File

@ -1,52 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import "errors"
type serverConfig struct {
handshaker Handshaker
interceptor UnaryServerInterceptor
}
// ServerOpt for configuring a ttrpc server
type ServerOpt func(*serverConfig) error
// WithServerHandshaker can be passed to NewServer to ensure that the
// handshaker is called before every connection attempt.
//
// Only one handshaker is allowed per server.
func WithServerHandshaker(handshaker Handshaker) ServerOpt {
return func(c *serverConfig) error {
if c.handshaker != nil {
return errors.New("only one handshaker allowed per server")
}
c.handshaker = handshaker
return nil
}
}
// WithUnaryServerInterceptor sets the provided interceptor on the server
func WithUnaryServerInterceptor(i UnaryServerInterceptor) ServerOpt {
return func(c *serverConfig) error {
if c.interceptor != nil {
return errors.New("only one interceptor allowed per server")
}
c.interceptor = i
return nil
}
}

View File

@ -1,23 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc defines and implements a low level simple transfer protocol
optimized for low latency and reliable connections between processes on the same
host. The protocol uses simple framing for sending requests, responses, and data
using multiple streams.
*/
package ttrpc

View File

@ -1,34 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import "errors"
var (
// ErrProtocol is a general error in the handling the protocol.
ErrProtocol = errors.New("protocol error")
// ErrClosed is returned by client methods when the underlying connection is
// closed.
ErrClosed = errors.New("ttrpc: closed")
// ErrServerClosed is returned when the Server has closed its connection.
ErrServerClosed = errors.New("ttrpc: server closed")
// ErrStreamClosed is when the streaming connection is closed.
ErrStreamClosed = errors.New("ttrpc: stream closed")
)

View File

@ -1,50 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"context"
"net"
)
// Handshaker defines the interface for connection handshakes performed on the
// server or client when first connecting.
type Handshaker interface {
// Handshake should confirm or decorate a connection that may be incoming
// to a server or outgoing from a client.
//
// If this returns without an error, the caller should use the connection
// in place of the original connection.
//
// The second return value can contain credential specific data, such as
// unix socket credentials or TLS information.
//
// While we currently only have implementations on the server-side, this
// interface should be sufficient to implement similar handshakes on the
// client-side.
Handshake(ctx context.Context, conn net.Conn) (net.Conn, interface{}, error)
}
type handshakerFunc func(ctx context.Context, conn net.Conn) (net.Conn, interface{}, error)
func (fn handshakerFunc) Handshake(ctx context.Context, conn net.Conn) (net.Conn, interface{}, error) {
return fn(ctx, conn)
}
func noopHandshake(_ context.Context, conn net.Conn) (net.Conn, interface{}, error) {
return conn, nil, nil
}

View File

@ -1,65 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import "context"
// UnaryServerInfo provides information about the server request
type UnaryServerInfo struct {
FullMethod string
}
// UnaryClientInfo provides information about the client request
type UnaryClientInfo struct {
FullMethod string
}
// StreamServerInfo provides information about the server request
type StreamServerInfo struct {
FullMethod string
StreamingClient bool
StreamingServer bool
}
// Unmarshaler contains the server request data and allows it to be unmarshaled
// into a concrete type
type Unmarshaler func(interface{}) error
// Invoker invokes the client's request and response from the ttrpc server
type Invoker func(context.Context, *Request, *Response) error
// UnaryServerInterceptor specifies the interceptor function for server request/response
type UnaryServerInterceptor func(context.Context, Unmarshaler, *UnaryServerInfo, Method) (interface{}, error)
// UnaryClientInterceptor specifies the interceptor function for client request/response
type UnaryClientInterceptor func(context.Context, *Request, *Response, *UnaryClientInfo, Invoker) error
func defaultServerInterceptor(ctx context.Context, unmarshal Unmarshaler, _ *UnaryServerInfo, method Method) (interface{}, error) {
return method(ctx, unmarshal)
}
func defaultClientInterceptor(ctx context.Context, req *Request, resp *Response, _ *UnaryClientInfo, invoker Invoker) error {
return invoker(ctx, req, resp)
}
type StreamServerInterceptor func(context.Context, StreamServer, *StreamServerInfo, StreamHandler) (interface{}, error)
func defaultStreamServerInterceptor(ctx context.Context, ss StreamServer, _ *StreamServerInfo, stream StreamHandler) (interface{}, error) {
return stream(ctx, ss)
}
type StreamClientInterceptor func(context.Context)

View File

@ -1,107 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"context"
"strings"
)
// MD is the user type for ttrpc metadata
type MD map[string][]string
// Get returns the metadata for a given key when they exist.
// If there is no metadata, a nil slice and false are returned.
func (m MD) Get(key string) ([]string, bool) {
key = strings.ToLower(key)
list, ok := m[key]
if !ok || len(list) == 0 {
return nil, false
}
return list, true
}
// Set sets the provided values for a given key.
// The values will overwrite any existing values.
// If no values provided, a key will be deleted.
func (m MD) Set(key string, values ...string) {
key = strings.ToLower(key)
if len(values) == 0 {
delete(m, key)
return
}
m[key] = values
}
// Append appends additional values to the given key.
func (m MD) Append(key string, values ...string) {
key = strings.ToLower(key)
if len(values) == 0 {
return
}
current, ok := m[key]
if ok {
m.Set(key, append(current, values...)...)
} else {
m.Set(key, values...)
}
}
func (m MD) setRequest(r *Request) {
for k, values := range m {
for _, v := range values {
r.Metadata = append(r.Metadata, &KeyValue{
Key: k,
Value: v,
})
}
}
}
func (m MD) fromRequest(r *Request) {
for _, kv := range r.Metadata {
m[kv.Key] = append(m[kv.Key], kv.Value)
}
}
type metadataKey struct{}
// GetMetadata retrieves metadata from context.Context (previously attached with WithMetadata)
func GetMetadata(ctx context.Context) (MD, bool) {
metadata, ok := ctx.Value(metadataKey{}).(MD)
return metadata, ok
}
// GetMetadataValue gets a specific metadata value by name from context.Context
func GetMetadataValue(ctx context.Context, name string) (string, bool) {
metadata, ok := GetMetadata(ctx)
if !ok {
return "", false
}
if list, ok := metadata.Get(name); ok {
return list[0], true
}
return "", false
}
// WithMetadata attaches metadata map to a context.Context
func WithMetadata(ctx context.Context, md MD) context.Context {
return context.WithValue(ctx, metadataKey{}, md)
}

View File

@ -1,396 +0,0 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.28.1
// protoc v3.20.1
// source: github.com/containerd/ttrpc/request.proto
package ttrpc
import (
status "google.golang.org/genproto/googleapis/rpc/status"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
type Request struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Service string `protobuf:"bytes,1,opt,name=service,proto3" json:"service,omitempty"`
Method string `protobuf:"bytes,2,opt,name=method,proto3" json:"method,omitempty"`
Payload []byte `protobuf:"bytes,3,opt,name=payload,proto3" json:"payload,omitempty"`
TimeoutNano int64 `protobuf:"varint,4,opt,name=timeout_nano,json=timeoutNano,proto3" json:"timeout_nano,omitempty"`
Metadata []*KeyValue `protobuf:"bytes,5,rep,name=metadata,proto3" json:"metadata,omitempty"`
}
func (x *Request) Reset() {
*x = Request{}
if protoimpl.UnsafeEnabled {
mi := &file_github_com_containerd_ttrpc_request_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *Request) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Request) ProtoMessage() {}
func (x *Request) ProtoReflect() protoreflect.Message {
mi := &file_github_com_containerd_ttrpc_request_proto_msgTypes[0]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Request.ProtoReflect.Descriptor instead.
func (*Request) Descriptor() ([]byte, []int) {
return file_github_com_containerd_ttrpc_request_proto_rawDescGZIP(), []int{0}
}
func (x *Request) GetService() string {
if x != nil {
return x.Service
}
return ""
}
func (x *Request) GetMethod() string {
if x != nil {
return x.Method
}
return ""
}
func (x *Request) GetPayload() []byte {
if x != nil {
return x.Payload
}
return nil
}
func (x *Request) GetTimeoutNano() int64 {
if x != nil {
return x.TimeoutNano
}
return 0
}
func (x *Request) GetMetadata() []*KeyValue {
if x != nil {
return x.Metadata
}
return nil
}
type Response struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Status *status.Status `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"`
}
func (x *Response) Reset() {
*x = Response{}
if protoimpl.UnsafeEnabled {
mi := &file_github_com_containerd_ttrpc_request_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *Response) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Response) ProtoMessage() {}
func (x *Response) ProtoReflect() protoreflect.Message {
mi := &file_github_com_containerd_ttrpc_request_proto_msgTypes[1]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Response.ProtoReflect.Descriptor instead.
func (*Response) Descriptor() ([]byte, []int) {
return file_github_com_containerd_ttrpc_request_proto_rawDescGZIP(), []int{1}
}
func (x *Response) GetStatus() *status.Status {
if x != nil {
return x.Status
}
return nil
}
func (x *Response) GetPayload() []byte {
if x != nil {
return x.Payload
}
return nil
}
type StringList struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
List []string `protobuf:"bytes,1,rep,name=list,proto3" json:"list,omitempty"`
}
func (x *StringList) Reset() {
*x = StringList{}
if protoimpl.UnsafeEnabled {
mi := &file_github_com_containerd_ttrpc_request_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *StringList) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*StringList) ProtoMessage() {}
func (x *StringList) ProtoReflect() protoreflect.Message {
mi := &file_github_com_containerd_ttrpc_request_proto_msgTypes[2]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use StringList.ProtoReflect.Descriptor instead.
func (*StringList) Descriptor() ([]byte, []int) {
return file_github_com_containerd_ttrpc_request_proto_rawDescGZIP(), []int{2}
}
func (x *StringList) GetList() []string {
if x != nil {
return x.List
}
return nil
}
type KeyValue struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
}
func (x *KeyValue) Reset() {
*x = KeyValue{}
if protoimpl.UnsafeEnabled {
mi := &file_github_com_containerd_ttrpc_request_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
}
func (x *KeyValue) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*KeyValue) ProtoMessage() {}
func (x *KeyValue) ProtoReflect() protoreflect.Message {
mi := &file_github_com_containerd_ttrpc_request_proto_msgTypes[3]
if protoimpl.UnsafeEnabled && x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use KeyValue.ProtoReflect.Descriptor instead.
func (*KeyValue) Descriptor() ([]byte, []int) {
return file_github_com_containerd_ttrpc_request_proto_rawDescGZIP(), []int{3}
}
func (x *KeyValue) GetKey() string {
if x != nil {
return x.Key
}
return ""
}
func (x *KeyValue) GetValue() string {
if x != nil {
return x.Value
}
return ""
}
var File_github_com_containerd_ttrpc_request_proto protoreflect.FileDescriptor
var file_github_com_containerd_ttrpc_request_proto_rawDesc = []byte{
0x0a, 0x29, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x6f, 0x6e,
0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x64, 0x2f, 0x74, 0x74, 0x72, 0x70, 0x63, 0x2f, 0x72, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x05, 0x74, 0x74, 0x72,
0x70, 0x63, 0x1a, 0x12, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73,
0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xa5, 0x01, 0x0a, 0x07, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x12, 0x18, 0x0a, 0x07, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x18, 0x01, 0x20,
0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x16, 0x0a, 0x06,
0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x6d, 0x65,
0x74, 0x68, 0x6f, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18,
0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x12, 0x21,
0x0a, 0x0c, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x5f, 0x6e, 0x61, 0x6e, 0x6f, 0x18, 0x04,
0x20, 0x01, 0x28, 0x03, 0x52, 0x0b, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x4e, 0x61, 0x6e,
0x6f, 0x12, 0x2b, 0x0a, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x18, 0x05, 0x20,
0x03, 0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x74, 0x74, 0x72, 0x70, 0x63, 0x2e, 0x4b, 0x65, 0x79, 0x56,
0x61, 0x6c, 0x75, 0x65, 0x52, 0x08, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x22, 0x45,
0x0a, 0x08, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1f, 0x0a, 0x06, 0x73, 0x74,
0x61, 0x74, 0x75, 0x73, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x53, 0x74, 0x61,
0x74, 0x75, 0x73, 0x52, 0x06, 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x18, 0x0a, 0x07, 0x70,
0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x70, 0x61,
0x79, 0x6c, 0x6f, 0x61, 0x64, 0x22, 0x20, 0x0a, 0x0a, 0x53, 0x74, 0x72, 0x69, 0x6e, 0x67, 0x4c,
0x69, 0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x6c, 0x69, 0x73, 0x74, 0x18, 0x01, 0x20, 0x03, 0x28,
0x09, 0x52, 0x04, 0x6c, 0x69, 0x73, 0x74, 0x22, 0x32, 0x0a, 0x08, 0x4b, 0x65, 0x79, 0x56, 0x61,
0x6c, 0x75, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02,
0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x42, 0x1d, 0x5a, 0x1b, 0x67,
0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x61, 0x69,
0x6e, 0x65, 0x72, 0x64, 0x2f, 0x74, 0x74, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74,
0x6f, 0x33,
}
var (
file_github_com_containerd_ttrpc_request_proto_rawDescOnce sync.Once
file_github_com_containerd_ttrpc_request_proto_rawDescData = file_github_com_containerd_ttrpc_request_proto_rawDesc
)
func file_github_com_containerd_ttrpc_request_proto_rawDescGZIP() []byte {
file_github_com_containerd_ttrpc_request_proto_rawDescOnce.Do(func() {
file_github_com_containerd_ttrpc_request_proto_rawDescData = protoimpl.X.CompressGZIP(file_github_com_containerd_ttrpc_request_proto_rawDescData)
})
return file_github_com_containerd_ttrpc_request_proto_rawDescData
}
var file_github_com_containerd_ttrpc_request_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
var file_github_com_containerd_ttrpc_request_proto_goTypes = []interface{}{
(*Request)(nil), // 0: ttrpc.Request
(*Response)(nil), // 1: ttrpc.Response
(*StringList)(nil), // 2: ttrpc.StringList
(*KeyValue)(nil), // 3: ttrpc.KeyValue
(*status.Status)(nil), // 4: Status
}
var file_github_com_containerd_ttrpc_request_proto_depIdxs = []int32{
3, // 0: ttrpc.Request.metadata:type_name -> ttrpc.KeyValue
4, // 1: ttrpc.Response.status:type_name -> Status
2, // [2:2] is the sub-list for method output_type
2, // [2:2] is the sub-list for method input_type
2, // [2:2] is the sub-list for extension type_name
2, // [2:2] is the sub-list for extension extendee
0, // [0:2] is the sub-list for field type_name
}
func init() { file_github_com_containerd_ttrpc_request_proto_init() }
func file_github_com_containerd_ttrpc_request_proto_init() {
if File_github_com_containerd_ttrpc_request_proto != nil {
return
}
if !protoimpl.UnsafeEnabled {
file_github_com_containerd_ttrpc_request_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Request); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_github_com_containerd_ttrpc_request_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*Response); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_github_com_containerd_ttrpc_request_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*StringList); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
file_github_com_containerd_ttrpc_request_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
switch v := v.(*KeyValue); i {
case 0:
return &v.state
case 1:
return &v.sizeCache
case 2:
return &v.unknownFields
default:
return nil
}
}
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_github_com_containerd_ttrpc_request_proto_rawDesc,
NumEnums: 0,
NumMessages: 4,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_github_com_containerd_ttrpc_request_proto_goTypes,
DependencyIndexes: file_github_com_containerd_ttrpc_request_proto_depIdxs,
MessageInfos: file_github_com_containerd_ttrpc_request_proto_msgTypes,
}.Build()
File_github_com_containerd_ttrpc_request_proto = out.File
file_github_com_containerd_ttrpc_request_proto_rawDesc = nil
file_github_com_containerd_ttrpc_request_proto_goTypes = nil
file_github_com_containerd_ttrpc_request_proto_depIdxs = nil
}

View File

@ -1,29 +0,0 @@
syntax = "proto3";
package ttrpc;
import "proto/status.proto";
option go_package = "github.com/containerd/ttrpc";
message Request {
string service = 1;
string method = 2;
bytes payload = 3;
int64 timeout_nano = 4;
repeated KeyValue metadata = 5;
}
message Response {
Status status = 1;
bytes payload = 2;
}
message StringList {
repeated string list = 1;
}
message KeyValue {
string key = 1;
string value = 2;
}

View File

@ -1,579 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"context"
"errors"
"io"
"math/rand"
"net"
"sync"
"sync/atomic"
"syscall"
"time"
"github.com/sirupsen/logrus"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
type Server struct {
config *serverConfig
services *serviceSet
codec codec
mu sync.Mutex
listeners map[net.Listener]struct{}
connections map[*serverConn]struct{} // all connections to current state
done chan struct{} // marks point at which we stop serving requests
}
func NewServer(opts ...ServerOpt) (*Server, error) {
config := &serverConfig{}
for _, opt := range opts {
if err := opt(config); err != nil {
return nil, err
}
}
if config.interceptor == nil {
config.interceptor = defaultServerInterceptor
}
return &Server{
config: config,
services: newServiceSet(config.interceptor),
done: make(chan struct{}),
listeners: make(map[net.Listener]struct{}),
connections: make(map[*serverConn]struct{}),
}, nil
}
// Register registers a map of methods to method handlers
// TODO: Remove in 2.0, does not support streams
func (s *Server) Register(name string, methods map[string]Method) {
s.services.register(name, &ServiceDesc{Methods: methods})
}
func (s *Server) RegisterService(name string, desc *ServiceDesc) {
s.services.register(name, desc)
}
func (s *Server) Serve(ctx context.Context, l net.Listener) error {
s.addListener(l)
defer s.closeListener(l)
var (
backoff time.Duration
handshaker = s.config.handshaker
)
if handshaker == nil {
handshaker = handshakerFunc(noopHandshake)
}
for {
conn, err := l.Accept()
if err != nil {
select {
case <-s.done:
return ErrServerClosed
default:
}
if terr, ok := err.(interface {
Temporary() bool
}); ok && terr.Temporary() {
if backoff == 0 {
backoff = time.Millisecond
} else {
backoff *= 2
}
if max := time.Second; backoff > max {
backoff = max
}
sleep := time.Duration(rand.Int63n(int64(backoff)))
logrus.WithError(err).Errorf("ttrpc: failed accept; backoff %v", sleep)
time.Sleep(sleep)
continue
}
return err
}
backoff = 0
approved, handshake, err := handshaker.Handshake(ctx, conn)
if err != nil {
logrus.WithError(err).Error("ttrpc: refusing connection after handshake")
conn.Close()
continue
}
sc, err := s.newConn(approved, handshake)
if err != nil {
logrus.WithError(err).Error("ttrpc: create connection failed")
conn.Close()
continue
}
go sc.run(ctx)
}
}
func (s *Server) Shutdown(ctx context.Context) error {
s.mu.Lock()
select {
case <-s.done:
default:
// protected by mutex
close(s.done)
}
lnerr := s.closeListeners()
s.mu.Unlock()
ticker := time.NewTicker(200 * time.Millisecond)
defer ticker.Stop()
for {
s.closeIdleConns()
if s.countConnection() == 0 {
break
}
select {
case <-ctx.Done():
return ctx.Err()
case <-ticker.C:
}
}
return lnerr
}
// Close the server without waiting for active connections.
func (s *Server) Close() error {
s.mu.Lock()
defer s.mu.Unlock()
select {
case <-s.done:
default:
// protected by mutex
close(s.done)
}
err := s.closeListeners()
for c := range s.connections {
c.close()
delete(s.connections, c)
}
return err
}
func (s *Server) addListener(l net.Listener) {
s.mu.Lock()
defer s.mu.Unlock()
s.listeners[l] = struct{}{}
}
func (s *Server) closeListener(l net.Listener) error {
s.mu.Lock()
defer s.mu.Unlock()
return s.closeListenerLocked(l)
}
func (s *Server) closeListenerLocked(l net.Listener) error {
defer delete(s.listeners, l)
return l.Close()
}
func (s *Server) closeListeners() error {
var err error
for l := range s.listeners {
if cerr := s.closeListenerLocked(l); cerr != nil && err == nil {
err = cerr
}
}
return err
}
func (s *Server) addConnection(c *serverConn) error {
s.mu.Lock()
defer s.mu.Unlock()
select {
case <-s.done:
return ErrServerClosed
default:
}
s.connections[c] = struct{}{}
return nil
}
func (s *Server) delConnection(c *serverConn) {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.connections, c)
}
func (s *Server) countConnection() int {
s.mu.Lock()
defer s.mu.Unlock()
return len(s.connections)
}
func (s *Server) closeIdleConns() {
s.mu.Lock()
defer s.mu.Unlock()
for c := range s.connections {
if st, ok := c.getState(); !ok || st == connStateActive {
continue
}
c.close()
delete(s.connections, c)
}
}
type connState int
const (
connStateActive = iota + 1 // outstanding requests
connStateIdle // no requests
connStateClosed // closed connection
)
func (cs connState) String() string {
switch cs {
case connStateActive:
return "active"
case connStateIdle:
return "idle"
case connStateClosed:
return "closed"
default:
return "unknown"
}
}
func (s *Server) newConn(conn net.Conn, handshake interface{}) (*serverConn, error) {
c := &serverConn{
server: s,
conn: conn,
handshake: handshake,
shutdown: make(chan struct{}),
}
c.setState(connStateIdle)
if err := s.addConnection(c); err != nil {
c.close()
return nil, err
}
return c, nil
}
type serverConn struct {
server *Server
conn net.Conn
handshake interface{} // data from handshake, not used for now
state atomic.Value
shutdownOnce sync.Once
shutdown chan struct{} // forced shutdown, used by close
}
func (c *serverConn) getState() (connState, bool) {
cs, ok := c.state.Load().(connState)
return cs, ok
}
func (c *serverConn) setState(newstate connState) {
c.state.Store(newstate)
}
func (c *serverConn) close() error {
c.shutdownOnce.Do(func() {
close(c.shutdown)
})
return nil
}
func (c *serverConn) run(sctx context.Context) {
type (
response struct {
id uint32
status *status.Status
data []byte
closeStream bool
streaming bool
}
)
var (
ch = newChannel(c.conn)
ctx, cancel = context.WithCancel(sctx)
state connState = connStateIdle
responses = make(chan response)
recvErr = make(chan error, 1)
done = make(chan struct{})
streams = sync.Map{}
active int32
lastStreamID uint32
)
defer c.conn.Close()
defer cancel()
defer close(done)
defer c.server.delConnection(c)
sendStatus := func(id uint32, st *status.Status) bool {
select {
case responses <- response{
// even though we've had an invalid stream id, we send it
// back on the same stream id so the client knows which
// stream id was bad.
id: id,
status: st,
closeStream: true,
}:
return true
case <-c.shutdown:
return false
case <-done:
return false
}
}
go func(recvErr chan error) {
defer close(recvErr)
for {
select {
case <-c.shutdown:
return
case <-done:
return
default: // proceed
}
mh, p, err := ch.recv()
if err != nil {
status, ok := status.FromError(err)
if !ok {
recvErr <- err
return
}
// in this case, we send an error for that particular message
// when the status is defined.
if !sendStatus(mh.StreamID, status) {
return
}
continue
}
if mh.StreamID%2 != 1 {
// enforce odd client initiated identifiers.
if !sendStatus(mh.StreamID, status.Newf(codes.InvalidArgument, "StreamID must be odd for client initiated streams")) {
return
}
continue
}
if mh.Type == messageTypeData {
i, ok := streams.Load(mh.StreamID)
if !ok {
if !sendStatus(mh.StreamID, status.Newf(codes.InvalidArgument, "StreamID is no longer active")) {
return
}
}
sh := i.(*streamHandler)
if mh.Flags&flagNoData != flagNoData {
unmarshal := func(obj interface{}) error {
err := protoUnmarshal(p, obj)
ch.putmbuf(p)
return err
}
if err := sh.data(unmarshal); err != nil {
if !sendStatus(mh.StreamID, status.Newf(codes.InvalidArgument, "data handling error: %v", err)) {
return
}
}
}
if mh.Flags&flagRemoteClosed == flagRemoteClosed {
sh.closeSend()
if len(p) > 0 {
if !sendStatus(mh.StreamID, status.Newf(codes.InvalidArgument, "data close message cannot include data")) {
return
}
}
}
} else if mh.Type == messageTypeRequest {
if mh.StreamID <= lastStreamID {
// enforce odd client initiated identifiers.
if !sendStatus(mh.StreamID, status.Newf(codes.InvalidArgument, "StreamID cannot be re-used and must increment")) {
return
}
continue
}
lastStreamID = mh.StreamID
// TODO: Make request type configurable
// Unmarshaller which takes in a byte array and returns an interface?
var req Request
if err := c.server.codec.Unmarshal(p, &req); err != nil {
ch.putmbuf(p)
if !sendStatus(mh.StreamID, status.Newf(codes.InvalidArgument, "unmarshal request error: %v", err)) {
return
}
continue
}
ch.putmbuf(p)
id := mh.StreamID
respond := func(status *status.Status, data []byte, streaming, closeStream bool) error {
select {
case responses <- response{
id: id,
status: status,
data: data,
closeStream: closeStream,
streaming: streaming,
}:
case <-done:
return ErrClosed
}
return nil
}
sh, err := c.server.services.handle(ctx, &req, respond)
if err != nil {
status, _ := status.FromError(err)
if !sendStatus(mh.StreamID, status) {
return
}
continue
}
streams.Store(id, sh)
atomic.AddInt32(&active, 1)
}
// TODO: else we must ignore this for future compat. log this?
}
}(recvErr)
for {
var (
newstate connState
shutdown chan struct{}
)
activeN := atomic.LoadInt32(&active)
if activeN > 0 {
newstate = connStateActive
shutdown = nil
} else {
newstate = connStateIdle
shutdown = c.shutdown // only enable this branch in idle mode
}
if newstate != state {
c.setState(newstate)
state = newstate
}
select {
case response := <-responses:
if !response.streaming || response.status.Code() != codes.OK {
p, err := c.server.codec.Marshal(&Response{
Status: response.status.Proto(),
Payload: response.data,
})
if err != nil {
logrus.WithError(err).Error("failed marshaling response")
return
}
if err := ch.send(response.id, messageTypeResponse, 0, p); err != nil {
logrus.WithError(err).Error("failed sending message on channel")
return
}
} else {
var flags uint8
if response.closeStream {
flags = flagRemoteClosed
}
if response.data == nil {
flags = flags | flagNoData
}
if err := ch.send(response.id, messageTypeData, flags, response.data); err != nil {
logrus.WithError(err).Error("failed sending message on channel")
return
}
}
if response.closeStream {
// The ttrpc protocol currently does not support the case where
// the server is localClosed but not remoteClosed. Once the server
// is closing, the whole stream may be considered finished
streams.Delete(response.id)
atomic.AddInt32(&active, -1)
}
case err := <-recvErr:
// TODO(stevvooe): Not wildly clear what we should do in this
// branch. Basically, it means that we are no longer receiving
// requests due to a terminal error.
recvErr = nil // connection is now "closing"
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, syscall.ECONNRESET) {
// The client went away and we should stop processing
// requests, so that the client connection is closed
return
}
logrus.WithError(err).Error("error receiving message")
// else, initiate shutdown
case <-shutdown:
return
}
}
}
var noopFunc = func() {}
func getRequestContext(ctx context.Context, req *Request) (retCtx context.Context, cancel func()) {
if len(req.Metadata) > 0 {
md := MD{}
md.fromRequest(req)
ctx = WithMetadata(ctx, md)
}
cancel = noopFunc
if req.TimeoutNano == 0 {
return ctx, cancel
}
ctx, cancel = context.WithTimeout(ctx, time.Duration(req.TimeoutNano))
return ctx, cancel
}

View File

@ -1,275 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"context"
"errors"
"fmt"
"io"
"os"
"path"
"unsafe"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
)
type Method func(ctx context.Context, unmarshal func(interface{}) error) (interface{}, error)
type StreamHandler func(context.Context, StreamServer) (interface{}, error)
type Stream struct {
Handler StreamHandler
StreamingClient bool
StreamingServer bool
}
type ServiceDesc struct {
Methods map[string]Method
Streams map[string]Stream
}
type serviceSet struct {
services map[string]*ServiceDesc
unaryInterceptor UnaryServerInterceptor
streamInterceptor StreamServerInterceptor
}
func newServiceSet(interceptor UnaryServerInterceptor) *serviceSet {
return &serviceSet{
services: make(map[string]*ServiceDesc),
unaryInterceptor: interceptor,
streamInterceptor: defaultStreamServerInterceptor,
}
}
func (s *serviceSet) register(name string, desc *ServiceDesc) {
if _, ok := s.services[name]; ok {
panic(fmt.Errorf("duplicate service %v registered", name))
}
s.services[name] = desc
}
func (s *serviceSet) unaryCall(ctx context.Context, method Method, info *UnaryServerInfo, data []byte) (p []byte, st *status.Status) {
unmarshal := func(obj interface{}) error {
return protoUnmarshal(data, obj)
}
resp, err := s.unaryInterceptor(ctx, unmarshal, info, method)
if err == nil {
if isNil(resp) {
err = errors.New("ttrpc: marshal called with nil")
} else {
p, err = protoMarshal(resp)
}
}
st, ok := status.FromError(err)
if !ok {
st = status.New(convertCode(err), err.Error())
}
return p, st
}
func (s *serviceSet) streamCall(ctx context.Context, stream StreamHandler, info *StreamServerInfo, ss StreamServer) (p []byte, st *status.Status) {
resp, err := s.streamInterceptor(ctx, ss, info, stream)
if err == nil {
p, err = protoMarshal(resp)
}
st, ok := status.FromError(err)
if !ok {
st = status.New(convertCode(err), err.Error())
}
return
}
func (s *serviceSet) handle(ctx context.Context, req *Request, respond func(*status.Status, []byte, bool, bool) error) (*streamHandler, error) {
srv, ok := s.services[req.Service]
if !ok {
return nil, status.Errorf(codes.Unimplemented, "service %v", req.Service)
}
if method, ok := srv.Methods[req.Method]; ok {
go func() {
ctx, cancel := getRequestContext(ctx, req)
defer cancel()
info := &UnaryServerInfo{
FullMethod: fullPath(req.Service, req.Method),
}
p, st := s.unaryCall(ctx, method, info, req.Payload)
respond(st, p, false, true)
}()
return nil, nil
}
if stream, ok := srv.Streams[req.Method]; ok {
ctx, cancel := getRequestContext(ctx, req)
info := &StreamServerInfo{
FullMethod: fullPath(req.Service, req.Method),
StreamingClient: stream.StreamingClient,
StreamingServer: stream.StreamingServer,
}
sh := &streamHandler{
ctx: ctx,
respond: respond,
recv: make(chan Unmarshaler, 5),
info: info,
}
go func() {
defer cancel()
p, st := s.streamCall(ctx, stream.Handler, info, sh)
respond(st, p, stream.StreamingServer, true)
}()
if req.Payload != nil {
unmarshal := func(obj interface{}) error {
return protoUnmarshal(req.Payload, obj)
}
if err := sh.data(unmarshal); err != nil {
return nil, err
}
}
return sh, nil
}
return nil, status.Errorf(codes.Unimplemented, "method %v", req.Method)
}
type streamHandler struct {
ctx context.Context
respond func(*status.Status, []byte, bool, bool) error
recv chan Unmarshaler
info *StreamServerInfo
remoteClosed bool
localClosed bool
}
func (s *streamHandler) closeSend() {
if !s.remoteClosed {
s.remoteClosed = true
close(s.recv)
}
}
func (s *streamHandler) data(unmarshal Unmarshaler) error {
if s.remoteClosed {
return ErrStreamClosed
}
select {
case s.recv <- unmarshal:
return nil
case <-s.ctx.Done():
return s.ctx.Err()
}
}
func (s *streamHandler) SendMsg(m interface{}) error {
if s.localClosed {
return ErrStreamClosed
}
p, err := protoMarshal(m)
if err != nil {
return err
}
return s.respond(nil, p, true, false)
}
func (s *streamHandler) RecvMsg(m interface{}) error {
select {
case unmarshal, ok := <-s.recv:
if !ok {
return io.EOF
}
return unmarshal(m)
case <-s.ctx.Done():
return s.ctx.Err()
}
}
func protoUnmarshal(p []byte, obj interface{}) error {
switch v := obj.(type) {
case proto.Message:
if err := proto.Unmarshal(p, v); err != nil {
return status.Errorf(codes.Internal, "ttrpc: error unmarshalling payload: %v", err.Error())
}
default:
return status.Errorf(codes.Internal, "ttrpc: error unsupported request type: %T", v)
}
return nil
}
func protoMarshal(obj interface{}) ([]byte, error) {
if obj == nil {
return nil, nil
}
switch v := obj.(type) {
case proto.Message:
r, err := proto.Marshal(v)
if err != nil {
return nil, status.Errorf(codes.Internal, "ttrpc: error marshaling payload: %v", err.Error())
}
return r, nil
default:
return nil, status.Errorf(codes.Internal, "ttrpc: error unsupported response type: %T", v)
}
}
// convertCode maps stdlib go errors into grpc space.
//
// This is ripped from the grpc-go code base.
func convertCode(err error) codes.Code {
switch err {
case nil:
return codes.OK
case io.EOF:
return codes.OutOfRange
case io.ErrClosedPipe, io.ErrNoProgress, io.ErrShortBuffer, io.ErrShortWrite, io.ErrUnexpectedEOF:
return codes.FailedPrecondition
case os.ErrInvalid:
return codes.InvalidArgument
case context.Canceled:
return codes.Canceled
case context.DeadlineExceeded:
return codes.DeadlineExceeded
}
switch {
case os.IsExist(err):
return codes.AlreadyExists
case os.IsNotExist(err):
return codes.NotFound
case os.IsPermission(err):
return codes.PermissionDenied
}
return codes.Unknown
}
func fullPath(service, method string) string {
return "/" + path.Join(service, method)
}
func isNil(resp interface{}) bool {
return (*[2]uintptr)(unsafe.Pointer(&resp))[1] == 0
}

View File

@ -1,84 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"context"
"sync"
)
type streamID uint32
type streamMessage struct {
header messageHeader
payload []byte
}
type stream struct {
id streamID
sender sender
recv chan *streamMessage
closeOnce sync.Once
recvErr error
recvClose chan struct{}
}
func newStream(id streamID, send sender) *stream {
return &stream{
id: id,
sender: send,
recv: make(chan *streamMessage, 1),
recvClose: make(chan struct{}),
}
}
func (s *stream) closeWithError(err error) error {
s.closeOnce.Do(func() {
if err != nil {
s.recvErr = err
} else {
s.recvErr = ErrClosed
}
close(s.recvClose)
})
return nil
}
func (s *stream) send(mt messageType, flags uint8, b []byte) error {
return s.sender.send(uint32(s.id), mt, flags, b)
}
func (s *stream) receive(ctx context.Context, msg *streamMessage) error {
select {
case <-s.recvClose:
return s.recvErr
default:
}
select {
case <-s.recvClose:
return s.recvErr
case s.recv <- msg:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
type sender interface {
send(uint32, messageType, uint8, []byte) error
}

View File

@ -1,22 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
type StreamServer interface {
SendMsg(m interface{}) error
RecvMsg(m interface{}) error
}

View File

@ -1,16 +0,0 @@
syntax = "proto3";
package ttrpc;
option go_package = "github.com/containerd/ttrpc/internal";
message TestPayload {
string foo = 1;
int64 deadline = 2;
string metadata = 3;
}
message EchoPayload {
int64 seq = 1;
string msg = 2;
}

View File

@ -1,105 +0,0 @@
/*
Copyright The containerd Authors.
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 ttrpc
import (
"context"
"errors"
"fmt"
"net"
"os"
"syscall"
"golang.org/x/sys/unix"
)
type UnixCredentialsFunc func(*unix.Ucred) error
func (fn UnixCredentialsFunc) Handshake(_ context.Context, conn net.Conn) (net.Conn, interface{}, error) {
uc, err := requireUnixSocket(conn)
if err != nil {
return nil, nil, fmt.Errorf("ttrpc.UnixCredentialsFunc: require unix socket: %w", err)
}
rs, err := uc.SyscallConn()
if err != nil {
return nil, nil, fmt.Errorf("ttrpc.UnixCredentialsFunc: (net.UnixConn).SyscallConn failed: %w", err)
}
var (
ucred *unix.Ucred
ucredErr error
)
if err := rs.Control(func(fd uintptr) {
ucred, ucredErr = unix.GetsockoptUcred(int(fd), unix.SOL_SOCKET, unix.SO_PEERCRED)
}); err != nil {
return nil, nil, fmt.Errorf("ttrpc.UnixCredentialsFunc: (*syscall.RawConn).Control failed: %w", err)
}
if ucredErr != nil {
return nil, nil, fmt.Errorf("ttrpc.UnixCredentialsFunc: failed to retrieve socket peer credentials: %w", ucredErr)
}
if err := fn(ucred); err != nil {
return nil, nil, fmt.Errorf("ttrpc.UnixCredentialsFunc: credential check failed: %w", err)
}
return uc, ucred, nil
}
// UnixSocketRequireUidGid requires specific *effective* UID/GID, rather than the real UID/GID.
//
// For example, if a daemon binary is owned by the root (UID 0) with SUID bit but running as an
// unprivileged user (UID 1001), the effective UID becomes 0, and the real UID becomes 1001.
// So calling this function with uid=0 allows a connection from effective UID 0 but rejects
// a connection from effective UID 1001.
//
// See socket(7), SO_PEERCRED: "The returned credentials are those that were in effect at the time of the call to connect(2) or socketpair(2)."
func UnixSocketRequireUidGid(uid, gid int) UnixCredentialsFunc {
return func(ucred *unix.Ucred) error {
return requireUidGid(ucred, uid, gid)
}
}
func UnixSocketRequireRoot() UnixCredentialsFunc {
return UnixSocketRequireUidGid(0, 0)
}
// UnixSocketRequireSameUser resolves the current effective unix user and returns a
// UnixCredentialsFunc that will validate incoming unix connections against the
// current credentials.
//
// This is useful when using abstract sockets that are accessible by all users.
func UnixSocketRequireSameUser() UnixCredentialsFunc {
euid, egid := os.Geteuid(), os.Getegid()
return UnixSocketRequireUidGid(euid, egid)
}
func requireUidGid(ucred *unix.Ucred, uid, gid int) error {
if (uid != -1 && uint32(uid) != ucred.Uid) || (gid != -1 && uint32(gid) != ucred.Gid) {
return fmt.Errorf("ttrpc: invalid credentials: %v", syscall.EPERM)
}
return nil
}
func requireUnixSocket(conn net.Conn) (*net.UnixConn, error) {
uc, ok := conn.(*net.UnixConn)
if !ok {
return nil, errors.New("a unix socket connection is required")
}
return uc, nil
}

View File

@ -1,237 +0,0 @@
package imageutil
import (
"context"
"encoding/json"
"sync"
"time"
"github.com/containerd/containerd/content"
"github.com/containerd/containerd/images"
"github.com/containerd/containerd/leases"
"github.com/containerd/containerd/platforms"
"github.com/containerd/containerd/reference"
"github.com/containerd/containerd/remotes"
"github.com/containerd/containerd/remotes/docker"
intoto "github.com/in-toto/in-toto-golang/in_toto"
"github.com/moby/buildkit/util/contentutil"
"github.com/moby/buildkit/util/leaseutil"
"github.com/moby/buildkit/util/resolver/limited"
"github.com/moby/buildkit/util/resolver/retryhandler"
digest "github.com/opencontainers/go-digest"
ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
)
type ContentCache interface {
content.Ingester
content.Provider
content.Manager
}
var leasesMu sync.Mutex
var leasesF []func(context.Context) error
func CancelCacheLeases() {
leasesMu.Lock()
for _, f := range leasesF {
f(context.TODO())
}
leasesF = nil
leasesMu.Unlock()
}
func AddLease(f func(context.Context) error) {
leasesMu.Lock()
leasesF = append(leasesF, f)
leasesMu.Unlock()
}
func Config(ctx context.Context, str string, resolver remotes.Resolver, cache ContentCache, leaseManager leases.Manager, p *ocispecs.Platform) (digest.Digest, []byte, error) {
// TODO: fix buildkit to take interface instead of struct
var platform platforms.MatchComparer
if p != nil {
platform = platforms.Only(*p)
} else {
platform = platforms.Default()
}
ref, err := reference.Parse(str)
if err != nil {
return "", nil, errors.WithStack(err)
}
if leaseManager != nil {
ctx2, done, err := leaseutil.WithLease(ctx, leaseManager, leases.WithExpiration(5*time.Minute), leaseutil.MakeTemporary)
if err != nil {
return "", nil, errors.WithStack(err)
}
ctx = ctx2
defer func() {
// this lease is not deleted to allow other components to access manifest/config from cache. It will be deleted after 5 min deadline or on pruning inactive builder
AddLease(done)
}()
}
desc := ocispecs.Descriptor{
Digest: ref.Digest(),
}
if desc.Digest != "" {
ra, err := cache.ReaderAt(ctx, desc)
if err == nil {
info, err := cache.Info(ctx, desc.Digest)
if err == nil {
if ok, err := contentutil.HasSource(info, ref); err == nil && ok {
desc.Size = ra.Size()
mt, err := DetectManifestMediaType(ra)
if err == nil {
desc.MediaType = mt
}
}
}
}
}
// use resolver if desc is incomplete
if desc.MediaType == "" {
_, desc, err = resolver.Resolve(ctx, ref.String())
if err != nil {
return "", nil, err
}
}
fetcher, err := resolver.Fetcher(ctx, ref.String())
if err != nil {
return "", nil, err
}
if desc.MediaType == images.MediaTypeDockerSchema1Manifest {
return readSchema1Config(ctx, ref.String(), desc, fetcher, cache)
}
children := childrenConfigHandler(cache, platform)
dslHandler, err := docker.AppendDistributionSourceLabel(cache, ref.String())
if err != nil {
return "", nil, err
}
handlers := []images.Handler{
retryhandler.New(limited.FetchHandler(cache, fetcher, str), func(_ []byte) {}),
dslHandler,
children,
}
if err := images.Dispatch(ctx, images.Handlers(handlers...), nil, desc); err != nil {
return "", nil, err
}
config, err := images.Config(ctx, cache, desc, platform)
if err != nil {
return "", nil, err
}
dt, err := content.ReadBlob(ctx, cache, config)
if err != nil {
return "", nil, err
}
return desc.Digest, dt, nil
}
func childrenConfigHandler(provider content.Provider, platform platforms.MatchComparer) images.HandlerFunc {
return func(ctx context.Context, desc ocispecs.Descriptor) ([]ocispecs.Descriptor, error) {
var descs []ocispecs.Descriptor
switch desc.MediaType {
case images.MediaTypeDockerSchema2Manifest, ocispecs.MediaTypeImageManifest:
p, err := content.ReadBlob(ctx, provider, desc)
if err != nil {
return nil, err
}
// TODO(stevvooe): We just assume oci manifest, for now. There may be
// subtle differences from the docker version.
var manifest ocispecs.Manifest
if err := json.Unmarshal(p, &manifest); err != nil {
return nil, err
}
descs = append(descs, manifest.Config)
case images.MediaTypeDockerSchema2ManifestList, ocispecs.MediaTypeImageIndex:
p, err := content.ReadBlob(ctx, provider, desc)
if err != nil {
return nil, err
}
var index ocispecs.Index
if err := json.Unmarshal(p, &index); err != nil {
return nil, err
}
if platform != nil {
for _, d := range index.Manifests {
if d.Platform == nil || platform.Match(*d.Platform) {
descs = append(descs, d)
}
}
} else {
descs = append(descs, index.Manifests...)
}
case images.MediaTypeDockerSchema2Config, ocispecs.MediaTypeImageConfig, docker.LegacyConfigMediaType,
intoto.PayloadType:
// childless data types.
return nil, nil
default:
return nil, errors.Errorf("encountered unknown type %v; children may not be fetched", desc.MediaType)
}
return descs, nil
}
}
// specs.MediaTypeImageManifest, // TODO: detect schema1/manifest-list
func DetectManifestMediaType(ra content.ReaderAt) (string, error) {
// TODO: schema1
dt := make([]byte, ra.Size())
if _, err := ra.ReadAt(dt, 0); err != nil {
return "", err
}
return DetectManifestBlobMediaType(dt)
}
func DetectManifestBlobMediaType(dt []byte) (string, error) {
var mfst struct {
MediaType *string `json:"mediaType"`
Config json.RawMessage `json:"config"`
Manifests json.RawMessage `json:"manifests"`
Layers json.RawMessage `json:"layers"`
}
if err := json.Unmarshal(dt, &mfst); err != nil {
return "", err
}
mt := images.MediaTypeDockerSchema2ManifestList
if mfst.Config != nil || mfst.Layers != nil {
mt = images.MediaTypeDockerSchema2Manifest
if mfst.Manifests != nil {
return "", errors.Errorf("invalid ambiguous manifest and manifest list")
}
}
if mfst.MediaType != nil {
switch *mfst.MediaType {
case images.MediaTypeDockerSchema2ManifestList, ocispecs.MediaTypeImageIndex:
if mt != images.MediaTypeDockerSchema2ManifestList {
return "", errors.Errorf("mediaType in manifest does not match manifest contents")
}
mt = *mfst.MediaType
case images.MediaTypeDockerSchema2Manifest, ocispecs.MediaTypeImageManifest:
if mt != images.MediaTypeDockerSchema2Manifest {
return "", errors.Errorf("mediaType in manifest does not match manifest contents")
}
mt = *mfst.MediaType
}
}
return mt, nil
}

View File

@ -1,88 +0,0 @@
package imageutil
import (
"context"
"encoding/json"
"io"
"strings"
"time"
"github.com/containerd/containerd/remotes"
"github.com/moby/buildkit/exporter/containerimage/image"
digest "github.com/opencontainers/go-digest"
ocispecs "github.com/opencontainers/image-spec/specs-go/v1"
"github.com/pkg/errors"
)
func readSchema1Config(ctx context.Context, ref string, desc ocispecs.Descriptor, fetcher remotes.Fetcher, cache ContentCache) (digest.Digest, []byte, error) {
rc, err := fetcher.Fetch(ctx, desc)
if err != nil {
return "", nil, err
}
defer rc.Close()
dt, err := io.ReadAll(rc)
if err != nil {
return "", nil, errors.Wrap(err, "failed to fetch schema1 manifest")
}
dt, err = convertSchema1ConfigMeta(dt)
if err != nil {
return "", nil, err
}
return desc.Digest, dt, nil
}
func convertSchema1ConfigMeta(in []byte) ([]byte, error) {
type history struct {
V1Compatibility string `json:"v1Compatibility"`
}
var m struct {
History []history `json:"history"`
}
if err := json.Unmarshal(in, &m); err != nil {
return nil, errors.Wrap(err, "failed to unmarshal schema1 manifest")
}
if len(m.History) == 0 {
return nil, errors.Errorf("invalid schema1 manifest")
}
var img image.Image
if err := json.Unmarshal([]byte(m.History[0].V1Compatibility), &img); err != nil {
return nil, errors.Wrap(err, "failed to unmarshal image from schema 1 history")
}
img.RootFS = ocispecs.RootFS{
Type: "layers", // filled in by exporter
}
img.History = make([]ocispecs.History, len(m.History))
for i := range m.History {
var h v1History
if err := json.Unmarshal([]byte(m.History[i].V1Compatibility), &h); err != nil {
return nil, errors.Wrap(err, "failed to unmarshal history")
}
img.History[len(m.History)-i-1] = ocispecs.History{
Author: h.Author,
Comment: h.Comment,
Created: &h.Created,
CreatedBy: strings.Join(h.ContainerConfig.Cmd, " "),
EmptyLayer: (h.ThrowAway != nil && *h.ThrowAway) || (h.Size != nil && *h.Size == 0),
}
}
dt, err := json.MarshalIndent(img, "", " ")
if err != nil {
return nil, errors.Wrap(err, "failed to marshal schema1 config")
}
return dt, nil
}
type v1History struct {
Author string `json:"author,omitempty"`
Created time.Time `json:"created"`
Comment string `json:"comment,omitempty"`
ThrowAway *bool `json:"throwaway,omitempty"`
Size *int `json:"Size,omitempty"` // used before ThrowAway field
ContainerConfig struct {
Cmd []string `json:"Cmd,omitempty"`
} `json:"container_config,omitempty"`
}

View File

@ -1,83 +0,0 @@
package leaseutil
import (
"context"
"time"
"github.com/containerd/containerd/leases"
"github.com/containerd/containerd/namespaces"
)
func WithLease(ctx context.Context, ls leases.Manager, opts ...leases.Opt) (context.Context, func(context.Context) error, error) {
_, ok := leases.FromContext(ctx)
if ok {
return ctx, func(context.Context) error {
return nil
}, nil
}
l, err := ls.Create(ctx, append([]leases.Opt{leases.WithRandomID(), leases.WithExpiration(time.Hour)}, opts...)...)
if err != nil {
return nil, nil, err
}
ctx = leases.WithLease(ctx, l.ID)
return ctx, func(ctx context.Context) error {
return ls.Delete(ctx, l)
}, nil
}
func MakeTemporary(l *leases.Lease) error {
if l.Labels == nil {
l.Labels = map[string]string{}
}
l.Labels["buildkit/lease.temporary"] = time.Now().UTC().Format(time.RFC3339Nano)
return nil
}
func WithNamespace(lm leases.Manager, ns string) *Manager {
return &Manager{manager: lm, ns: ns}
}
type Manager struct {
manager leases.Manager
ns string
}
func (l *Manager) Namespace() string {
return l.ns
}
func (l *Manager) WithNamespace(ns string) *Manager {
return WithNamespace(l.manager, ns)
}
func (l *Manager) Create(ctx context.Context, opts ...leases.Opt) (leases.Lease, error) {
ctx = namespaces.WithNamespace(ctx, l.ns)
return l.manager.Create(ctx, opts...)
}
func (l *Manager) Delete(ctx context.Context, lease leases.Lease, opts ...leases.DeleteOpt) error {
ctx = namespaces.WithNamespace(ctx, l.ns)
return l.manager.Delete(ctx, lease, opts...)
}
func (l *Manager) List(ctx context.Context, filters ...string) ([]leases.Lease, error) {
ctx = namespaces.WithNamespace(ctx, l.ns)
return l.manager.List(ctx, filters...)
}
func (l *Manager) AddResource(ctx context.Context, lease leases.Lease, resource leases.Resource) error {
ctx = namespaces.WithNamespace(ctx, l.ns)
return l.manager.AddResource(ctx, lease, resource)
}
func (l *Manager) DeleteResource(ctx context.Context, lease leases.Lease, resource leases.Resource) error {
ctx = namespaces.WithNamespace(ctx, l.ns)
return l.manager.DeleteResource(ctx, lease, resource)
}
func (l *Manager) ListResources(ctx context.Context, lease leases.Lease) ([]leases.Resource, error) {
ctx = namespaces.WithNamespace(ctx, l.ns)
return l.manager.ListResources(ctx, lease)
}

8
vendor/modules.txt vendored
View File

@ -157,13 +157,10 @@ github.com/containerd/containerd/content/proxy
github.com/containerd/containerd/defaults
github.com/containerd/containerd/errdefs
github.com/containerd/containerd/filters
github.com/containerd/containerd/identifiers
github.com/containerd/containerd/images
github.com/containerd/containerd/images/archive
github.com/containerd/containerd/labels
github.com/containerd/containerd/leases
github.com/containerd/containerd/log
github.com/containerd/containerd/namespaces
github.com/containerd/containerd/pkg/dialer
github.com/containerd/containerd/pkg/randutil
github.com/containerd/containerd/pkg/seed
@ -191,9 +188,6 @@ github.com/containerd/continuity/fs/fstest
github.com/containerd/continuity/pathdriver
github.com/containerd/continuity/proto
github.com/containerd/continuity/sysx
# github.com/containerd/ttrpc v1.2.2
## explicit; go 1.13
github.com/containerd/ttrpc
# github.com/containerd/typeurl/v2 v2.1.1
## explicit; go 1.13
github.com/containerd/typeurl/v2
@ -553,8 +547,6 @@ github.com/moby/buildkit/util/entitlements
github.com/moby/buildkit/util/flightcontrol
github.com/moby/buildkit/util/gitutil
github.com/moby/buildkit/util/grpcerrors
github.com/moby/buildkit/util/imageutil
github.com/moby/buildkit/util/leaseutil
github.com/moby/buildkit/util/progress
github.com/moby/buildkit/util/progress/progressui
github.com/moby/buildkit/util/progress/progresswriter