Get ClientAssign partially working end-to-end.

Still has a lot of unimplemented corner cases, but it's now possible to
get a server assignment under some circumstances.

Also:

* PROTO ABI BREAK: swap Event fields 2 and 3 (location and unique)
* PROTO ABI BREAK: change Event field 8 (weight) from float to double
* PROTO API BREAK: rename [has_]shard_id → [has_]shard_number
* PROTO API BREAK: rename unique → unique_id
* PROTO API BREAK: rename Event weight → assigned_cost_per_second
* GO API BREAK: rename ATCUniqueError → ATCUniqueIDError
* GO API BREAK: rename ValidateATCUnique → ValidateATCUniqueID
* GO API BREAK: rename Resolved Unique → UniqueID
* GO API BREAK: rename Resolved [Has]ShardID → [Has]ShardNumber
* GO API BREAK: rename ATCTarget Unique → UniqueID
* GO API BREAK: rename ATCTarget [Has]ShardID → [Has]ShardNumber
* GO API BREAK: move mainutil SetUniqueFile/UniqueID to atcclient
* GO API BREAK: rename membership ServerSet/Roxy [Has]ShardID → [Has]ShardNumber
* GO API BREAK: delete roxyresolver GetShardID/WithShardID
* GO API BREAK: rename mainutil [Etcd|ZK]AnnounceConfig Unique → HostID
* GO API BREAK: rename mainutil ATCAnnounceConfig Unique → UniqueID
* GO API BREAK: rename atcclient Key [Has]ShardID → [Has]ShardNumber
* JSON BREAK: rename membership gRPC metadata key "ShardID" → "Shard"
* JSON BREAK: rename membership Roxy JSON key "shardID" → "shard"
* DATA BREAK: rename mainutil [Etcd|ZK]AnnounceConfig "unique" → "hostID"
* DATA BREAK: rename mainutil ATCAnnounceConfig "unique" → "uniqueID"
* Fix roxyresolver WatchingResolver and "atc" scheme deadlocks
* Fix bug where ATCClient ServerAnnounce/ClientAssign would forget to
  CloseSend, leading to hang
* ATCClient now falls back on SetUniqueFile/UniqueID
* InterceptorFactory cost-tallying interceptors now ignore all ATC RPCs
  and all healthcheck RPCs
* Fix bug in demo-backend processing of HTTP HEAD requests
* ATC does better sanity checking of ServerAnnounce inputs
* ATC no longer hangs until dirty shutdown when there are outstanding
  ServerAnnounce/ClientAssign RPC streams
* ATC now sends an empty ServerAnnounceResponse after successful
  connect, resulting in a retry counter reset to 0 in ATCClient

Author: chronos-tachyon

.gitignore

@@ -4,6 +4,7 @@
 /atcctl
 /colorize-logs
 /demo-backend
+/demo-client
 /zkcp
 /*.deb
 /release-notes.txt

cmd/atc/server_clientassign.go

@@ -1,6 +1,11 @@
 package main
 
 import (
+	"context"
+	"errors"
+	"io"
+
+	"github.com/rs/zerolog"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
@@ -39,12 +44,17 @@ func (s *ATCServer) ClientAssign(
 		return status.Error(codes.InvalidArgument, "first is absent")
 	}
 
-	err = roxyutil.ValidateATCLocation(first.Location)
+	err = roxyutil.ValidateATCServiceName(first.ServiceName)
+	if err != nil {
+		return status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	err = roxyutil.ValidateATCUniqueID(first.UniqueId)
 	if err != nil {
 		return status.Error(codes.InvalidArgument, err.Error())
 	}
 
-	err = roxyutil.ValidateATCUnique(first.Unique)
+	err = roxyutil.ValidateATCLocation(first.Location)
 	if err != nil {
 		return status.Error(codes.InvalidArgument, err.Error())
 	}
@@ -57,7 +67,7 @@ func (s *ATCServer) ClientAssign(
 		}
 	}()
 
-	key, svc, err := impl.ServiceMap.CheckInput(first.ServiceName, first.ShardId, first.HasShardId, false)
+	key, svc, err := impl.ServiceMap.CheckInput(first.ServiceName, first.ShardNumber, first.HasShardNumber, false)
 
 	logger.Debug().
 		Stringer("key", key).
@@ -72,5 +82,170 @@ func (s *ATCServer) ClientAssign(
 		return err
 	}
 
-	return status.Errorf(codes.Unimplemented, "method ClientAssign not implemented")
+	shardData := s.ref.GetOrInsertShard(key, svc, impl.CostMap)
+
+	shardData.Mutex.Lock()
+	clientData := shardData.LockedGetOrInsertClient(first)
+	clientData.CostHistory.UpdateAbsolute(req.CostCounter)
+	clientData.LockedUpdate(true, req.IsServing)
+	clientData.LockedOnConnect()
+	goAwayCh := (<-chan *roxy_v0.GoAway)(clientData.GoAwayCh)
+	shardData.Mutex.Unlock()
+
+	needImplRelease = false
+	s.ref.ReleaseSharedImpl()
+
+	active := &activeClientAssign{
+		logger:     logger,
+		s:          s,
+		cas:        cas,
+		shardData:  shardData,
+		clientData: clientData,
+		flushCh:    clientData.FlushCh,
+		errCh:      make(chan error),
+	}
+
+	go active.recvThread(req.IsServing)
+
+	err = active.doSend(&roxy_v0.ClientAssignResponse{})
+	if err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case <-gMultiServer.Done():
+			err := status.Error(codes.Unavailable, "ATC server is shutting down")
+			logger.Trace().
+				Err(err).
+				Msg("Return")
+			return err
+
+		case err := <-active.errCh:
+			return err
+
+		case goAway, ok := <-goAwayCh:
+			if !ok {
+				return <-active.errCh
+			}
+
+			if goAway == nil {
+				err := status.Errorf(codes.NotFound, "Key %v no longer exists", shardData.Key)
+				logger.Trace().
+					Err(err).
+					Msg("Return")
+				return err
+			}
+
+			err := active.doSend(&roxy_v0.ClientAssignResponse{GoAway: goAway})
+			if err == nil {
+				err = <-active.errCh
+			}
+			return err
+
+		case <-active.flushCh:
+			shardData.Mutex.Lock()
+			events := clientData.EventQueue
+			clientData.EventQueue = nil
+			shardData.Mutex.Unlock()
+
+			if len(events) != 0 {
+				err := active.doSend(&roxy_v0.ClientAssignResponse{Events: events})
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+}
+
+type activeClientAssign struct {
+	logger     zerolog.Logger
+	s          *ATCServer
+	cas        roxy_v0.AirTrafficControl_ClientAssignServer
+	shardData  *ShardData
+	clientData *ClientData
+	flushCh    chan struct{}
+	errCh      chan error
+}
+
+func (active *activeClientAssign) doSend(resp *roxy_v0.ClientAssignResponse) error {
+	active.logger.Trace().
+		Str("func", "ClientAssign.Send").
+		Interface("resp", resp).
+		Msg("Call")
+
+	err := active.cas.Send(resp)
+	if err != nil {
+		active.logger.Error().
+			Str("func", "ClientAssign.Send").
+			Err(err).
+			Msg("Error")
+	}
+	return err
+}
+
+func (active *activeClientAssign) recvThread(lastIsServing bool) {
+	var err error
+	var sendErr bool
+	defer func() {
+		active.shardData.Mutex.Lock()
+		active.clientData.CostHistory.Update()
+		active.clientData.LockedUpdate(false, lastIsServing)
+		active.shardData.Mutex.Unlock()
+
+		if sendErr {
+			select {
+			case active.errCh <- err:
+			default:
+			}
+		}
+
+		close(active.errCh)
+	}()
+
+	for {
+		active.logger.Trace().
+			Str("func", "ClientAssign.Recv").
+			Msg("Call")
+
+		var req *roxy_v0.ClientAssignRequest
+		req, err = active.cas.Recv()
+		s, ok := status.FromError(err)
+
+		switch {
+		case err == nil:
+			active.logger.Trace().
+				Interface("req", req).
+				Msg("Request")
+
+		case err == io.EOF:
+			fallthrough
+		case errors.Is(err, context.Canceled):
+			fallthrough
+		case errors.Is(err, context.DeadlineExceeded):
+			fallthrough
+		case ok && s.Code() == codes.Canceled:
+			active.logger.Trace().
+				Str("func", "ClientAssign.Recv").
+				Err(err).
+				Msg("Hangup")
+			return
+
+		default:
+			active.logger.Error().
+				Str("func", "ClientAssign.Recv").
+				Err(err).
+				Msg("Error")
+			sendErr = true
+			return
+		}
+
+		active.shardData.Mutex.Lock()
+		active.clientData.CostHistory.UpdateAbsolute(req.CostCounter)
+		active.clientData.LockedUpdate(true, req.IsServing)
+		active.shardData.Mutex.Unlock()
+
+		lastIsServing = req.IsServing
+	}
 }

cmd/atc/server_find.go

@@ -22,7 +22,7 @@ func (s *ATCServer) Find(
 	impl := s.ref.AcquireSharedImpl()
 	defer s.ref.ReleaseSharedImpl()
 
-	key, _, err := impl.ServiceMap.CheckInput(req.ServiceName, req.ShardId, req.HasShardId, false)
+	key, _, err := impl.ServiceMap.CheckInput(req.ServiceName, req.ShardNumber, req.HasShardNumber, false)
 
 	logger.Debug().
 		Stringer("key", key).

cmd/atc/server_lookup.go

@@ -19,7 +19,7 @@ func (s *ATCServer) Lookup(
 	impl := s.ref.AcquireSharedImpl()
 	defer s.ref.ReleaseSharedImpl()
 
-	key, svc, err := impl.ServiceMap.CheckInput(req.ServiceName, req.ShardId, req.HasShardId, true)
+	key, svc, err := impl.ServiceMap.CheckInput(req.ServiceName, req.ShardNumber, req.HasShardNumber, true)
 
 	logger.Debug().
 		Stringer("key", key).
@@ -40,10 +40,10 @@ func (s *ATCServer) Lookup(
 		AvgDemandedCostPerQuery:           svc.AvgDemandedCPQ,
 	}
 
-	if svc.IsSharded && !req.HasShardId {
-		shardLimit := ShardID(svc.NumShards)
+	if svc.IsSharded && !req.HasShardNumber {
+		shardLimit := ShardNumber(svc.NumShards)
 		s.ref.mu.Lock()
-		for id := ShardID(0); id < shardLimit; id++ {
+		for id := ShardNumber(0); id < shardLimit; id++ {
 			key2 := Key{key.ServiceName, id, true}
 			shardData := s.ref.shardsByKey[key2]
 			if shardData != nil {
@@ -74,11 +74,11 @@ func (s *ATCServer) LookupClients(
 	impl := s.ref.AcquireSharedImpl()
 	defer s.ref.ReleaseSharedImpl()
 
-	key, _, err := impl.ServiceMap.CheckInput(req.ServiceName, req.ShardId, req.HasShardId, false)
+	key, _, err := impl.ServiceMap.CheckInput(req.ServiceName, req.ShardNumber, req.HasShardNumber, false)
 
 	logger.Debug().
 		Stringer("key", key).
-		Str("uniqueID", req.Unique).
+		Str("uniqueID", req.UniqueId).
 		Msg("RPC")
 
 	if err != nil {
@@ -90,14 +90,14 @@ func (s *ATCServer) LookupClients(
 	shardData := s.ref.Shard(key)
 	if shardData != nil {
 		shardData.Mutex.Lock()
-		if req.Unique == "" {
-			resp.Clients = make([]*roxy_v0.ClientData, 0, len(shardData.ClientsByUnique))
-			for _, clientData := range shardData.ClientsByUnique {
+		if req.UniqueId == "" {
+			resp.Clients = make([]*roxy_v0.ClientData, 0, len(shardData.Clients))
+			for _, clientData := range shardData.Clients {
 				resp.Clients = append(resp.Clients, clientData.LockedToProto())
 			}
 		} else {
 			resp.Clients = make([]*roxy_v0.ClientData, 0, 1)
-			clientData := shardData.ClientsByUnique[req.Unique]
+			clientData := shardData.Clients[req.UniqueId]
 			if clientData != nil {
 				resp.Clients = append(resp.Clients, clientData.LockedToProto())
 			}
@@ -121,11 +121,11 @@ func (s *ATCServer) LookupServers(
 	impl := s.ref.AcquireSharedImpl()
 	defer s.ref.ReleaseSharedImpl()
 
-	key, _, err := impl.ServiceMap.CheckInput(req.ServiceName, req.ShardId, req.HasShardId, false)
+	key, _, err := impl.ServiceMap.CheckInput(req.ServiceName, req.ShardNumber, req.HasShardNumber, false)
 
 	logger.Debug().
 		Stringer("key", key).
-		Str("uniqueID", req.Unique).
+		Str("uniqueID", req.UniqueId).
 		Msg("RPC")
 
 	if err != nil {
@@ -137,14 +137,14 @@ func (s *ATCServer) LookupServers(
 	shardData := s.ref.Shard(key)
 	if shardData != nil {
 		shardData.Mutex.Lock()
-		if req.Unique == "" {
-			resp.Servers = make([]*roxy_v0.ServerData, 0, len(shardData.ServersByUnique))
-			for _, serverData := range shardData.ServersByUnique {
+		if req.UniqueId == "" {
+			resp.Servers = make([]*roxy_v0.ServerData, 0, len(shardData.Servers))
+			for _, serverData := range shardData.Servers {
 				resp.Servers = append(resp.Servers, serverData.LockedToProto())
 			}
 		} else {
 			resp.Servers = make([]*roxy_v0.ServerData, 0, 1)
-			serverData := shardData.ServersByUnique[req.Unique]
+			serverData := shardData.Servers[req.UniqueId]
 			if serverData != nil {
 				resp.Servers = append(resp.Servers, serverData.LockedToProto())
 			}