mirror of
https://github.com/psviderski/uncloud.git
synced 2026-08-26 11:03:34 +00:00
fix: race on warned flag in versioncheck pkg, make consistent use of client/server terms
This commit is contained in:
Pasha Sviderski
@@ -0,0 +1,184 @@
|
||||
package grpcversion
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/Masterminds/semver"
|
||||
"github.com/psviderski/uncloud/internal/version"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
const (
|
||||
MetadataKeyClientVersion = "uncloud-client-version"
|
||||
MetadataKeyMinServerVersion = "uncloud-min-server-version"
|
||||
MetadataKeyServerVersion = "uncloud-server-version"
|
||||
|
||||
// MinClientVersion is the minimum client version the daemon accepts. The daemon
|
||||
// rejects requests from older clients, forcing them to upgrade. This provides
|
||||
// a clean cut-off for dropping support for old clients.
|
||||
//
|
||||
// MinServerVersion is the minimum daemon version the client requires. The client
|
||||
// sends this with each request so the daemon can immediately reject if it's too old,
|
||||
// avoiding the need for a preflight request. This is useful when a new client feature
|
||||
// requires daemon capabilities that didn't exist in older versions.
|
||||
//
|
||||
// The two minimums are independent: a client might require a newer daemon for new
|
||||
// features, while that same daemon could still handle requests from older clients.
|
||||
MinClientVersion = "0.0.0"
|
||||
MinServerVersion = "0.0.0"
|
||||
|
||||
ReleaseURL = "https://github.com/psviderski/uncloud/releases/latest"
|
||||
)
|
||||
|
||||
var (
|
||||
// currentVersion is the version of this binary (CLI or daemon).
|
||||
currentVersion = semver.MustParse(version.String())
|
||||
// zeroVersion is used when no version is specified (treated as 0.0.0).
|
||||
zeroVersion = semver.MustParse("0.0.0")
|
||||
// Pre-parsed minimum versions for comparison.
|
||||
minClientVersion = semver.MustParse(MinClientVersion)
|
||||
minServerVersion = semver.MustParse(MinServerVersion)
|
||||
|
||||
// warned tracks if we've already printed the daemon version warning.
|
||||
// TODO: Remove when checkServerVersionInResponse is no longer needed (see below).
|
||||
warned atomic.Bool
|
||||
|
||||
// WarnWriter is the writer used for version mismatch warnings. Defaults to os.Stderr.
|
||||
// Tests can override this to capture warning output.
|
||||
WarnWriter io.Writer = os.Stderr
|
||||
)
|
||||
|
||||
func extractVersion(md metadata.MD, key string) *semver.Version {
|
||||
if md == nil {
|
||||
return zeroVersion
|
||||
}
|
||||
values := md.Get(key)
|
||||
if len(values) == 0 || values[0] == "" {
|
||||
return zeroVersion
|
||||
}
|
||||
sv, err := semver.NewVersion(values[0])
|
||||
if err != nil {
|
||||
return zeroVersion
|
||||
}
|
||||
return sv
|
||||
}
|
||||
|
||||
func checkClientVersionHeaders(ctx context.Context) error {
|
||||
md, _ := metadata.FromIncomingContext(ctx)
|
||||
|
||||
actualClientVersion := extractVersion(md, MetadataKeyClientVersion)
|
||||
if actualClientVersion.LessThan(minClientVersion) {
|
||||
return status.Errorf(codes.FailedPrecondition,
|
||||
"version check failed: client version is below minimum %s. Please upgrade: %s",
|
||||
minClientVersion, ReleaseURL)
|
||||
}
|
||||
|
||||
requiredMinServer := extractVersion(md, MetadataKeyMinServerVersion)
|
||||
if currentVersion.LessThan(requiredMinServer) {
|
||||
return status.Errorf(codes.FailedPrecondition,
|
||||
"version check failed: daemon version %s is below client's minimum required version %s. Please upgrade the daemon: %s",
|
||||
currentVersion, requiredMinServer, ReleaseURL)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func ServerUnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
||||
if err := checkClientVersionHeaders(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := grpc.SetHeader(ctx, metadata.Pairs(MetadataKeyServerVersion, currentVersion.String())); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return handler(ctx, req)
|
||||
}
|
||||
|
||||
func ServerStreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
||||
if err := checkClientVersionHeaders(ss.Context()); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := ss.SetHeader(metadata.Pairs(MetadataKeyServerVersion, currentVersion.String())); err != nil {
|
||||
return err
|
||||
}
|
||||
return handler(srv, ss)
|
||||
}
|
||||
|
||||
func ClientUnaryInterceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
|
||||
ctx = metadata.AppendToOutgoingContext(ctx,
|
||||
MetadataKeyClientVersion, currentVersion.String(),
|
||||
MetadataKeyMinServerVersion, MinServerVersion,
|
||||
)
|
||||
|
||||
// TODO: Remove when checkServerVersionInResponse is no longer needed,
|
||||
// as we'll no longer need to extract headers from the response here.
|
||||
var respMD metadata.MD
|
||||
opts = append(opts, grpc.Header(&respMD))
|
||||
|
||||
err := invoker(ctx, method, req, reply, cc, opts...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// TODO: Remove eventually (see note on method below).
|
||||
checkServerVersionInResponse(respMD)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// checkServerVersionInResponse warns the user when they communicated with a daemon that
|
||||
// did not check the version requirements. This is only needed during the transition to
|
||||
// version-checking releases.
|
||||
// TODO: Remove this in some later release, after users have upgraded.
|
||||
func checkServerVersionInResponse(md metadata.MD) {
|
||||
serverVersion := extractVersion(md, MetadataKeyServerVersion)
|
||||
if serverVersion.LessThan(minServerVersion) {
|
||||
if warned.Swap(true) {
|
||||
return
|
||||
}
|
||||
|
||||
msg := fmt.Sprintf("daemon version is below minimum required version %s. The daemon did not verify this CLI's minimum version requirement, so the operation may not have behaved as intended. Please upgrade the daemon: %s",
|
||||
minServerVersion, ReleaseURL)
|
||||
fmt.Fprintf(WarnWriter, "WARNING: %s\n", msg)
|
||||
}
|
||||
}
|
||||
|
||||
func ClientStreamInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
|
||||
ctx = metadata.AppendToOutgoingContext(ctx,
|
||||
MetadataKeyClientVersion, currentVersion.String(),
|
||||
MetadataKeyMinServerVersion, MinServerVersion,
|
||||
)
|
||||
|
||||
stream, err := streamer(ctx, desc, cc, method, opts...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// TODO: Wrapping the stream in versionedClientStream will no longer
|
||||
// be necessary when we are ready to remove the temporary, transition
|
||||
// safety check checkServerVersionInResponse (see note on method above).
|
||||
return &versionedClientStream{ClientStream: stream}, nil
|
||||
}
|
||||
|
||||
// TODO: Remove when checkServerVersionInResponse is no longer needed.
|
||||
type versionedClientStream struct {
|
||||
grpc.ClientStream
|
||||
}
|
||||
|
||||
// TODO: Remove when checkServerVersionInResponse is no longer needed.
|
||||
func (s *versionedClientStream) Header() (metadata.MD, error) {
|
||||
md, err := s.ClientStream.Header()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
checkServerVersionInResponse(md)
|
||||
|
||||
return md, nil
|
||||
}
|
||||
@@ -0,0 +1,195 @@
|
||||
package grpcversion
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
func TestExtractVersion(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
md metadata.MD
|
||||
key string
|
||||
expected string
|
||||
}{
|
||||
{
|
||||
name: "nil metadata",
|
||||
md: nil,
|
||||
key: MetadataKeyClientVersion,
|
||||
expected: "0.0.0",
|
||||
},
|
||||
{
|
||||
name: "missing key",
|
||||
md: metadata.MD{},
|
||||
key: MetadataKeyClientVersion,
|
||||
expected: "0.0.0",
|
||||
},
|
||||
{
|
||||
name: "empty value",
|
||||
md: metadata.Pairs(MetadataKeyClientVersion, ""),
|
||||
key: MetadataKeyClientVersion,
|
||||
expected: "0.0.0",
|
||||
},
|
||||
{
|
||||
name: "invalid version",
|
||||
md: metadata.Pairs(MetadataKeyClientVersion, "not-a-version"),
|
||||
key: MetadataKeyClientVersion,
|
||||
expected: "0.0.0",
|
||||
},
|
||||
{
|
||||
name: "valid version",
|
||||
md: metadata.Pairs(MetadataKeyClientVersion, "1.2.3"),
|
||||
key: MetadataKeyClientVersion,
|
||||
expected: "1.2.3",
|
||||
},
|
||||
{
|
||||
name: "version with prerelease",
|
||||
md: metadata.Pairs(MetadataKeyClientVersion, "0.0.0-dev"),
|
||||
key: MetadataKeyClientVersion,
|
||||
expected: "0.0.0-dev",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := extractVersion(tt.md, tt.key)
|
||||
assert.Equal(t, tt.expected, got.String())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckClientVersionHeaders(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
md metadata.MD
|
||||
wantErr bool
|
||||
errCode codes.Code
|
||||
errContain string
|
||||
}{
|
||||
{
|
||||
name: "cli version below minimum",
|
||||
md: metadata.Pairs(
|
||||
MetadataKeyClientVersion, "0.0.0-dev",
|
||||
),
|
||||
wantErr: true,
|
||||
errCode: codes.FailedPrecondition,
|
||||
errContain: "client version is below minimum",
|
||||
},
|
||||
{
|
||||
name: "cli version above minimum",
|
||||
md: metadata.Pairs(
|
||||
MetadataKeyClientVersion, "999.0.0",
|
||||
),
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "min daemon version above current daemon",
|
||||
md: metadata.Pairs(
|
||||
MetadataKeyClientVersion, "999.0.0",
|
||||
MetadataKeyMinServerVersion, "999.0.0",
|
||||
),
|
||||
wantErr: true,
|
||||
errCode: codes.FailedPrecondition,
|
||||
errContain: "daemon version",
|
||||
},
|
||||
{
|
||||
name: "min daemon version below current daemon",
|
||||
md: metadata.Pairs(
|
||||
MetadataKeyClientVersion, "999.0.0",
|
||||
MetadataKeyMinServerVersion, "0.0.1",
|
||||
),
|
||||
wantErr: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
if tt.md != nil {
|
||||
ctx = metadata.NewIncomingContext(ctx, tt.md)
|
||||
}
|
||||
|
||||
err := checkClientVersionHeaders(ctx)
|
||||
|
||||
if tt.wantErr {
|
||||
require.Error(t, err)
|
||||
st, ok := status.FromError(err)
|
||||
require.True(t, ok, "expected gRPC status error, got %T", err)
|
||||
assert.Equal(t, tt.errCode, st.Code())
|
||||
assert.Contains(t, st.Message(), tt.errContain)
|
||||
} else {
|
||||
assert.NoError(t, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func captureWarnings(t *testing.T, fn func()) string {
|
||||
t.Helper()
|
||||
var buf bytes.Buffer
|
||||
old := WarnWriter
|
||||
WarnWriter = &buf
|
||||
t.Cleanup(func() { WarnWriter = old })
|
||||
fn()
|
||||
return buf.String()
|
||||
}
|
||||
|
||||
func TestCheckServerVersionInResponse(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
md metadata.MD
|
||||
wantWarning bool
|
||||
}{
|
||||
{
|
||||
name: "daemon version below minimum",
|
||||
md: metadata.Pairs(MetadataKeyServerVersion, "0.0.0-dev"),
|
||||
wantWarning: true,
|
||||
},
|
||||
{
|
||||
name: "daemon version above minimum",
|
||||
md: metadata.Pairs(MetadataKeyServerVersion, "999.0.0"),
|
||||
wantWarning: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
warned.Store(false)
|
||||
|
||||
output := captureWarnings(t, func() {
|
||||
checkServerVersionInResponse(tt.md)
|
||||
})
|
||||
|
||||
if tt.wantWarning {
|
||||
assert.Contains(t, output, "WARNING")
|
||||
} else {
|
||||
assert.Empty(t, output)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckServerVersionInResponse_WarnOnce(t *testing.T) {
|
||||
warned.Store(false)
|
||||
|
||||
md := metadata.Pairs(MetadataKeyServerVersion, "0.0.0-dev")
|
||||
|
||||
// First call should warn.
|
||||
output1 := captureWarnings(t, func() {
|
||||
checkServerVersionInResponse(md)
|
||||
})
|
||||
assert.Contains(t, output1, "WARNING", "first call should warn")
|
||||
|
||||
// Second call should not warn (warned flag is now true).
|
||||
output2 := captureWarnings(t, func() {
|
||||
checkServerVersionInResponse(md)
|
||||
})
|
||||
assert.Empty(t, output2, "second call should not warn")
|
||||
}
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
"github.com/psviderski/uncloud/internal/corrosion"
|
||||
"github.com/psviderski/uncloud/internal/docker"
|
||||
"github.com/psviderski/uncloud/internal/fs"
|
||||
"github.com/psviderski/uncloud/internal/grpcversion"
|
||||
"github.com/psviderski/uncloud/internal/journal"
|
||||
"github.com/psviderski/uncloud/internal/machine/api/pb"
|
||||
apiproxy "github.com/psviderski/uncloud/internal/machine/api/proxy"
|
||||
@@ -34,7 +35,6 @@ import (
|
||||
"github.com/psviderski/uncloud/internal/machine/network"
|
||||
"github.com/psviderski/uncloud/internal/machine/store"
|
||||
"github.com/psviderski/uncloud/pkg/api"
|
||||
versionpkg "github.com/psviderski/uncloud/pkg/versioncheck"
|
||||
"github.com/psviderski/unregistry"
|
||||
"github.com/siderolabs/grpc-proxy/proxy"
|
||||
"golang.org/x/sync/errgroup"
|
||||
@@ -273,8 +273,8 @@ func NewMachine(config *Config) (*Machine, error) {
|
||||
proxyDirector := apiproxy.NewDirector(config.MachineSockPath, constants.MachineAPIPort)
|
||||
localProxyServer := grpc.NewServer(
|
||||
grpc.ForceServerCodecV2(proxy.Codec()),
|
||||
grpc.UnaryInterceptor(versionpkg.ServerUnaryInterceptor),
|
||||
grpc.StreamInterceptor(versionpkg.ServerStreamInterceptor),
|
||||
grpc.UnaryInterceptor(grpcversion.ServerUnaryInterceptor),
|
||||
grpc.StreamInterceptor(grpcversion.ServerStreamInterceptor),
|
||||
grpc.UnknownServiceHandler(
|
||||
proxy.TransparentHandler(proxyDirector.Director),
|
||||
),
|
||||
@@ -428,8 +428,8 @@ func (m *Machine) Run(ctx context.Context) error {
|
||||
m.proxyDirector.UpdateLocalAddress(m.state.Network.ManagementIP.String())
|
||||
proxyServer := grpc.NewServer(
|
||||
grpc.ForceServerCodecV2(proxy.Codec()),
|
||||
grpc.UnaryInterceptor(versionpkg.ServerUnaryInterceptor),
|
||||
grpc.StreamInterceptor(versionpkg.ServerStreamInterceptor),
|
||||
grpc.UnaryInterceptor(grpcversion.ServerUnaryInterceptor),
|
||||
grpc.StreamInterceptor(grpcversion.ServerStreamInterceptor),
|
||||
grpc.UnknownServiceHandler(
|
||||
proxy.TransparentHandler(m.proxyDirector.Director),
|
||||
),
|
||||
|
||||
Reference in New Issue
Block a user