add starter code and readme for grpc expander usage
This commit is contained in:
Evan Sheng
parent
4504f55485
commit
a2b24e005f
|
|
@ -28,8 +28,6 @@ import (
|
|||
"k8s.io/autoscaler/cluster-autoscaler/expander/waste"
|
||||
"k8s.io/autoscaler/cluster-autoscaler/utils/errors"
|
||||
"k8s.io/autoscaler/cluster-autoscaler/utils/kubernetes"
|
||||
"k8s.io/klog/v2"
|
||||
|
||||
kube_client "k8s.io/client-go/kubernetes"
|
||||
)
|
||||
|
||||
|
|
@ -71,7 +69,6 @@ func ExpanderStrategyFromStrings(expanderFlags []string, cloudProvider cloudprov
|
|||
lister := kubernetes.NewConfigMapListerForNamespace(kubeClient, stopChannel, configNamespace)
|
||||
filters = append(filters, priority.NewFilter(lister.ConfigMaps(configNamespace), autoscalingKubeClients.Recorder))
|
||||
case expander.GRPCExpanderName:
|
||||
klog.V(1).Info("GRPC expander chosen")
|
||||
filters = append(filters, grpcplugin.NewFilter(GRPCExpanderCert, GRPCExpanderURL))
|
||||
default:
|
||||
return nil, errors.NewAutoscalerError(errors.InternalError, "Expander %s not supported", expanderFlag)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,41 @@
|
|||
# gRPC Expander for Cluster Autoscaler
|
||||
|
||||
## Introduction
|
||||
This expander functions as a gRPC client, and will pass expansion options to an external gRPC server.
|
||||
The external server will use this information to make a decision on which Node Group to expand, and return an option to expand.
|
||||
|
||||
## Motivation
|
||||
|
||||
This expander gives users very fine grained control over which option they'd like to expand.
|
||||
The gRPC server must be implemented by the user, but the logic can be developed out of band with Cluster Autoscaler.
|
||||
There are a wide variety of use cases here. Some examples are as follows:
|
||||
* A tiered weighted random strategy can be implemented, instead of a static priority ladder offered by the priority expander.
|
||||
* A strategy to encapsulate business logic specific to a user but not all users of Cluster Autoscaler
|
||||
* A strategy to take into account the dynamic fluctuating prices of the spot instance market
|
||||
|
||||
## Configuration options
|
||||
As using this expander requires communication with another service, users must specify a few options as CLI arguments.
|
||||
|
||||
```yaml
|
||||
--grpcExpanderUrl
|
||||
```
|
||||
URL of the gRPC Expander server, for CA to communicate with.
|
||||
```yaml
|
||||
--grpcExpanderCert
|
||||
```
|
||||
Location of the volume mounted certificate of the gRPC server if it is configured to communicate over TLS
|
||||
|
||||
## gRPC Expander Server Setup
|
||||
The gRPC server can be set up in many ways, but a simple example is described below.
|
||||
An example of a barebones gRPC Exapnder Server can be found in the `example` directory under `fake_grpc_server.go` file. This is meant to be copied elsewhere and deployed as a separate
|
||||
service. Note that the `protos/expander.pb.go` generated protobuf code will also need to be copied and used to serialize/deserizle the Options passed from CA.
|
||||
Communication between Cluster Autoscaler and the gRPC Server will occur over native kube-proxy. To use this, note the Service and Namespace the gRPC server is deployed in.
|
||||
|
||||
Deploy the gRPC Expander Server as a separate app, listening on a specifc port number.
|
||||
Start Cluster Autoscaler with the `--grpcExapnderURl=SERVICE_NAME.NAMESPACE_NAME.svc.cluster.local:PORT_NUMBER` flag, as well as `--grpcExpanderCert` pointed at the location of the volume mounted certificate of the gRPC server.
|
||||
|
||||
## Details
|
||||
|
||||
The gRPC client currently transforms nodeInfo objects passed into the expander to v1.Node objects to save rpc call throughput. As such, the gRPC server will not have access to daemonsets and static pods running on each node.
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,104 @@
|
|||
/*
|
||||
Copyright 2021 The Kubernetes 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 example
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"k8s.io/autoscaler/cluster-autoscaler/expander/grpcplugin/protos"
|
||||
)
|
||||
|
||||
// This code is meant to be used as starter code, deployed as a separate app, not in Cluster Autoscaler.
|
||||
// This serves as the gRPC Expander Server counterpart to the client which lives in this repo
|
||||
// main.go of said application should simply pass in paths to (optional)cert, (optional)private key, and port, and call Serve to start listening
|
||||
// copy the protos/expander.pb.go to your other application's repo, so it has access to the protobuf definitions
|
||||
|
||||
// Serve should be called by the main() function in main.go of the Expander Server repo to start serving
|
||||
func Serve(certPath string, keyPath string, port uint) {
|
||||
|
||||
var grpcServer *grpc.Server
|
||||
|
||||
// If credentials are passed in, use them
|
||||
if certPath != "" && keyPath != "" {
|
||||
log.Printf("Using certFile: %v and keyFile: %v", certPath, keyPath)
|
||||
tlsCredentials, err := credentials.NewServerTLSFromFile(certPath, keyPath)
|
||||
if err != nil {
|
||||
log.Fatal("cannot load TLS credentials: ", err)
|
||||
}
|
||||
grpcServer = grpc.NewServer(grpc.Creds(tlsCredentials))
|
||||
} else {
|
||||
grpcServer = grpc.NewServer()
|
||||
}
|
||||
|
||||
netListener := getNetListener(port)
|
||||
|
||||
expanderServerImpl := NewExpanderServerImpl()
|
||||
|
||||
protos.RegisterExpanderServer(grpcServer, expanderServerImpl)
|
||||
|
||||
// start the server
|
||||
log.Println("Starting server on port ", port)
|
||||
if err := grpcServer.Serve(netListener); err != nil {
|
||||
log.Fatalf("failed to serve: %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
func getNetListener(port uint) net.Listener {
|
||||
lis, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
||||
if err != nil {
|
||||
log.Fatalf("failed to listen: %v", err)
|
||||
panic(fmt.Sprintf("failed to listen: %v", err))
|
||||
}
|
||||
|
||||
return lis
|
||||
}
|
||||
|
||||
// ExpanderServerImpl is an implementation of Expander Server from proto definition
|
||||
type ExpanderServerImpl struct{}
|
||||
|
||||
// NewExpanderServerImpl is this Expander's implementation of the server
|
||||
func NewExpanderServerImpl() *ExpanderServerImpl {
|
||||
return &ExpanderServerImpl{}
|
||||
}
|
||||
|
||||
// BestOptions method filters out the best options of all options passed from the gRPC Client in CA, according to the defined strategy.
|
||||
func (ServerImpl *ExpanderServerImpl) BestOptions(ctx context.Context, req *protos.BestOptionsRequest) (*protos.BestOptionsResponse, error) {
|
||||
opts := req.GetOptions()
|
||||
log.Printf("Received BestOption Request with %v options", len(opts))
|
||||
|
||||
// This strategy simply chooses the Option with the longest NodeGroupID name, but can be replaced with any arbitrary logic
|
||||
longest := 0
|
||||
var choice *protos.Option
|
||||
for _, opt := range opts {
|
||||
log.Println(opt.NodeGroupId)
|
||||
if len(opt.NodeGroupId) > longest {
|
||||
choice = opt
|
||||
}
|
||||
}
|
||||
|
||||
log.Print("returned bestOptions with option: ", choice.NodeGroupId)
|
||||
|
||||
// Return just one option for now
|
||||
return &protos.BestOptionsResponse{
|
||||
Options: []*protos.Option{choice},
|
||||
}, nil
|
||||
}
|
||||
|
|
@ -0,0 +1,30 @@
|
|||
/*
|
||||
Copyright 2021 The Kubernetes 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 example
|
||||
|
||||
import "flag"
|
||||
|
||||
func main() {
|
||||
|
||||
certPath := flag.String("cert-path", "", "Path to cert file for gRPC Expander Server")
|
||||
keyPath := flag.String("key-path", "", "Path to private key for gRPC Expander Server")
|
||||
port := flag.Uint("port", 7000, "Port number for server to listen on")
|
||||
|
||||
flag.Parse()
|
||||
|
||||
Serve(*certPath, *keyPath, *port)
|
||||
}
|
||||
|
|
@ -31,6 +31,8 @@ import (
|
|||
"google.golang.org/grpc/credentials"
|
||||
)
|
||||
|
||||
const gRPCTimeout = 5 * time.Second
|
||||
|
||||
type grpcclientstrategy struct {
|
||||
grpcClient protos.ExpanderClient
|
||||
}
|
||||
|
|
@ -47,21 +49,20 @@ func NewFilter(expanderCert string, expanderUrl string) expander.Filter {
|
|||
func createGRPCClient(expanderCert string, expanderUrl string) protos.ExpanderClient {
|
||||
var dialOpt grpc.DialOption
|
||||
|
||||
// if no Cert file specified, use insecure
|
||||
if expanderCert == "" {
|
||||
dialOpt = grpc.WithInsecure()
|
||||
} else {
|
||||
creds, err := credentials.NewClientTLSFromFile(expanderCert, "")
|
||||
if err != nil {
|
||||
log.Fatalf("Failed to create TLS credentials %v", err)
|
||||
return nil
|
||||
}
|
||||
dialOpt = grpc.WithTransportCredentials(creds)
|
||||
log.Fatalf("GRPC Expander Cert not specified, insecure connections not allowed")
|
||||
return nil
|
||||
}
|
||||
klog.V(2).Info("Dialing ", expanderUrl, " dialopt: ", dialOpt)
|
||||
creds, err := credentials.NewClientTLSFromFile(expanderCert, "")
|
||||
if err != nil {
|
||||
log.Fatalf("Failed to create TLS credentials %v", err)
|
||||
return nil
|
||||
}
|
||||
dialOpt = grpc.WithTransportCredentials(creds)
|
||||
klog.V(2).Infof("Dialing: %s with dialopt: %v", expanderUrl, dialOpt)
|
||||
conn, err := grpc.Dial(expanderUrl, dialOpt)
|
||||
if err != nil {
|
||||
log.Fatalf("fail to dial server: %v", err)
|
||||
log.Fatalf("Fail to dial server: %v", err)
|
||||
return nil
|
||||
}
|
||||
return protos.NewExpanderClient(conn)
|
||||
|
|
@ -69,67 +70,69 @@ func createGRPCClient(expanderCert string, expanderUrl string) protos.ExpanderCl
|
|||
|
||||
func (g *grpcclientstrategy) BestOptions(expansionOptions []expander.Option, nodeInfo map[string]*schedulerframework.NodeInfo) []expander.Option {
|
||||
if g.grpcClient == nil {
|
||||
log.Fatalf("Incorrect gRPC client config, filtering no options")
|
||||
klog.Errorf("Incorrect gRPC client config, filtering no options")
|
||||
return expansionOptions
|
||||
}
|
||||
|
||||
// Transform inputs to gRPC inputs
|
||||
nodeGroupIDOptionMap := make(map[string]expander.Option)
|
||||
grpcOptionsSlice := []*protos.Option{}
|
||||
populateOptionsForGRPC(expansionOptions, nodeGroupIDOptionMap, &grpcOptionsSlice)
|
||||
grpcNodeInfoMap := make(map[string]*v1.Node)
|
||||
populateNodeInfoForGRPC(nodeInfo, grpcNodeInfoMap)
|
||||
grpcOptionsSlice, nodeGroupIDOptionMap := populateOptionsForGRPC(expansionOptions)
|
||||
grpcNodeMap := populateNodeInfoForGRPC(nodeInfo)
|
||||
|
||||
// call gRPC server to get BestOption
|
||||
klog.V(2).Info("GPRC call of best options to server with ", len(nodeGroupIDOptionMap), " options")
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
klog.V(2).Infof("GPRC call of best options to server with %v options", len(nodeGroupIDOptionMap))
|
||||
ctx, cancel := context.WithTimeout(context.Background(), gRPCTimeout)
|
||||
defer cancel()
|
||||
bestOptionsResponse, err := g.grpcClient.BestOptions(ctx, &protos.BestOptionsRequest{Options: grpcOptionsSlice, NodeInfoMap: grpcNodeInfoMap})
|
||||
bestOptionsResponse, err := g.grpcClient.BestOptions(ctx, &protos.BestOptionsRequest{Options: grpcOptionsSlice, NodeMap: grpcNodeMap})
|
||||
if err != nil {
|
||||
klog.V(2).Info("GRPC call timed out, no options filtered")
|
||||
klog.V(4).Info("GRPC call timed out, no options filtered")
|
||||
return expansionOptions
|
||||
}
|
||||
|
||||
if bestOptionsResponse == nil || bestOptionsResponse.Options == nil {
|
||||
klog.V(2).Info("GRPC returned nil bestOptions, no options filtered")
|
||||
klog.V(4).Info("GRPC returned nil bestOptions, no options filtered")
|
||||
return expansionOptions
|
||||
}
|
||||
// Transform back options slice
|
||||
options := transformAndSanitizeOptionsFromGRPC(bestOptionsResponse.Options, nodeGroupIDOptionMap)
|
||||
if options == nil {
|
||||
klog.V(2).Info("Unable to sanitize GPRC returned bestOptions, no options filtered")
|
||||
klog.V(4).Info("Unable to sanitize GPRC returned bestOptions, no options filtered")
|
||||
return expansionOptions
|
||||
}
|
||||
return options
|
||||
}
|
||||
|
||||
// populateOptionsForGRPC creates a map of nodegroup ID and options, as well as a slice of Options objects for the gRPC call
|
||||
func populateOptionsForGRPC(expansionOptions []expander.Option, nodeGroupIDOptionMap map[string]expander.Option, grpcOptionsSlice *[]*protos.Option) {
|
||||
func populateOptionsForGRPC(expansionOptions []expander.Option) ([]*protos.Option, map[string]expander.Option) {
|
||||
grpcOptionsSlice := []*protos.Option{}
|
||||
nodeGroupIDOptionMap := make(map[string]expander.Option)
|
||||
for _, option := range expansionOptions {
|
||||
nodeGroupIDOptionMap[option.NodeGroup.Id()] = option
|
||||
*grpcOptionsSlice = append(*grpcOptionsSlice, newOptionMessage(option.NodeGroup.Id(), int32(option.NodeCount), option.Debug, option.Pods))
|
||||
grpcOptionsSlice = append(grpcOptionsSlice, newOptionMessage(option.NodeGroup.Id(), int32(option.NodeCount), option.Debug, option.Pods))
|
||||
}
|
||||
return grpcOptionsSlice, nodeGroupIDOptionMap
|
||||
}
|
||||
|
||||
// populateNodeInfoForGRPC modifies the nodeInfo object, and replaces it with the v1.Node to pass through grpc
|
||||
func populateNodeInfoForGRPC(nodeInfos map[string]*schedulerframework.NodeInfo, grpcNodeInfoMap map[string]*v1.Node) {
|
||||
// populateNodeInfoForGRPC looks at the corresponding v1.Node object per NodeInfo object, and populates the grpcNodeInfoMap with these to pass over grpc
|
||||
func populateNodeInfoForGRPC(nodeInfos map[string]*schedulerframework.NodeInfo) map[string]*v1.Node {
|
||||
grpcNodeInfoMap := make(map[string]*v1.Node)
|
||||
for nodeId, nodeInfo := range nodeInfos {
|
||||
grpcNodeInfoMap[nodeId] = nodeInfo.Node()
|
||||
}
|
||||
return grpcNodeInfoMap
|
||||
}
|
||||
|
||||
func transformAndSanitizeOptionsFromGRPC(bestOptionsResponseOptions []*protos.Option, nodeGroupIDOptionMap map[string]expander.Option) []expander.Option {
|
||||
var options []expander.Option
|
||||
for _, option := range bestOptionsResponseOptions {
|
||||
if option == nil {
|
||||
klog.Errorf("gRPC server returned nil Option")
|
||||
return nil
|
||||
klog.Errorf("GRPC server returned nil Option")
|
||||
continue
|
||||
}
|
||||
if _, ok := nodeGroupIDOptionMap[option.NodeGroupId]; ok {
|
||||
options = append(options, nodeGroupIDOptionMap[option.NodeGroupId])
|
||||
} else {
|
||||
klog.Errorf("gRPC server returned invalid nodeGroup ID: ", option.NodeGroupId)
|
||||
return nil
|
||||
klog.Errorf("GRPC server returned invalid nodeGroup ID: ", option.NodeGroupId)
|
||||
continue
|
||||
}
|
||||
}
|
||||
return options
|
||||
|
|
|
|||
|
|
@ -87,20 +87,41 @@ var (
|
|||
)
|
||||
|
||||
func TestPopulateOptionsForGrpc(t *testing.T) {
|
||||
nodeGroupIDOptionMap := make(map[string]expander.Option)
|
||||
grpcOptionsSlice := []*protos.Option{}
|
||||
populateOptionsForGRPC(options, nodeGroupIDOptionMap, &grpcOptionsSlice)
|
||||
|
||||
expectedOptionsSlice := []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge}
|
||||
assert.Equal(t, expectedOptionsSlice, grpcOptionsSlice)
|
||||
|
||||
expectedNodeGroupIDOptionMap := map[string]expander.Option{
|
||||
eoT2Micro.NodeGroup.Id(): eoT2Micro,
|
||||
eoT2Large.NodeGroup.Id(): eoT2Large,
|
||||
eoT3Large.NodeGroup.Id(): eoT3Large,
|
||||
eoM44XLarge.NodeGroup.Id(): eoM44XLarge,
|
||||
testCases := []struct {
|
||||
desc string
|
||||
opts []expander.Option
|
||||
expectedOpts []*protos.Option
|
||||
expectedMap map[string]expander.Option
|
||||
}{
|
||||
{
|
||||
desc: "empty options",
|
||||
opts: []expander.Option{},
|
||||
expectedOpts: []*protos.Option{},
|
||||
expectedMap: map[string]expander.Option{},
|
||||
},
|
||||
{
|
||||
desc: "one option",
|
||||
opts: []expander.Option{eoT2Micro},
|
||||
expectedOpts: []*protos.Option{&grpcEoT2Micro},
|
||||
expectedMap: map[string]expander.Option{eoT2Micro.NodeGroup.Id(): eoT2Micro},
|
||||
},
|
||||
{
|
||||
desc: "many options",
|
||||
opts: options,
|
||||
expectedOpts: []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
|
||||
expectedMap: map[string]expander.Option{
|
||||
eoT2Micro.NodeGroup.Id(): eoT2Micro,
|
||||
eoT2Large.NodeGroup.Id(): eoT2Large,
|
||||
eoT3Large.NodeGroup.Id(): eoT3Large,
|
||||
eoM44XLarge.NodeGroup.Id(): eoM44XLarge,
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tc := range testCases {
|
||||
grpcOptionsSlice, nodeGroupIDOptionMap := populateOptionsForGRPC(tc.opts)
|
||||
assert.Equal(t, tc.expectedOpts, grpcOptionsSlice)
|
||||
assert.Equal(t, tc.expectedMap, nodeGroupIDOptionMap)
|
||||
}
|
||||
assert.Equal(t, expectedNodeGroupIDOptionMap, nodeGroupIDOptionMap)
|
||||
}
|
||||
|
||||
func makeFakeNodeInfos() map[string]*schedulerframework.NodeInfo {
|
||||
|
|
@ -114,9 +135,8 @@ func makeFakeNodeInfos() map[string]*schedulerframework.NodeInfo {
|
|||
}
|
||||
|
||||
func TestPopulateNodeInfoForGRPC(t *testing.T) {
|
||||
grpcNodeInfoMap := make(map[string]*v1.Node)
|
||||
nodeInfos := makeFakeNodeInfos()
|
||||
populateNodeInfoForGRPC(nodeInfos, grpcNodeInfoMap)
|
||||
grpcNodeInfoMap := populateNodeInfoForGRPC(nodeInfos)
|
||||
|
||||
expectedGrpcNodeInfoMap := make(map[string]*v1.Node)
|
||||
for i, opt := range options {
|
||||
|
|
@ -140,7 +160,7 @@ func TestValidTransformAndSanitizeOptionsFromGRPC(t *testing.T) {
|
|||
assert.Equal(t, expectedOptions, ret)
|
||||
}
|
||||
|
||||
func TestInvalidTransformAndSanitizeOptionsFromGRPC(t *testing.T) {
|
||||
func TestAnInvalidTransformAndSanitizeOptionsFromGRPC(t *testing.T) {
|
||||
responseOptionsSlice := []*protos.Option{&grpcEoT2Micro, &grpcEoT3Large, &grpcEoM44XLarge}
|
||||
nodeGroupIDOptionMap := map[string]expander.Option{
|
||||
eoT2Micro.NodeGroup.Id(): eoT2Micro,
|
||||
|
|
@ -149,7 +169,7 @@ func TestInvalidTransformAndSanitizeOptionsFromGRPC(t *testing.T) {
|
|||
}
|
||||
|
||||
ret := transformAndSanitizeOptionsFromGRPC(responseOptionsSlice, nodeGroupIDOptionMap)
|
||||
assert.Equal(t, []expander.Option(nil), ret)
|
||||
assert.Equal(t, []expander.Option{eoT2Micro, eoT3Large}, ret)
|
||||
}
|
||||
|
||||
func TestBestOptionsValid(t *testing.T) {
|
||||
|
|
@ -164,8 +184,8 @@ func TestBestOptionsValid(t *testing.T) {
|
|||
grpcNodeInfoMap[opt.NodeGroup.Id()] = nodes[i]
|
||||
}
|
||||
expectedBestOptionsReq := &protos.BestOptionsRequest{
|
||||
Options: []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
|
||||
NodeInfoMap: grpcNodeInfoMap,
|
||||
Options: []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
|
||||
NodeMap: grpcNodeInfoMap,
|
||||
}
|
||||
|
||||
mockClient.EXPECT().BestOptions(
|
||||
|
|
@ -242,13 +262,12 @@ func TestBestOptionsErrors(t *testing.T) {
|
|||
},
|
||||
}
|
||||
for _, tc := range testCases {
|
||||
grpcNodeInfoMap := make(map[string]*v1.Node)
|
||||
populateNodeInfoForGRPC(tc.nodeInfo, grpcNodeInfoMap)
|
||||
grpcNodeInfoMap := populateNodeInfoForGRPC(tc.nodeInfo)
|
||||
mockClient.EXPECT().BestOptions(
|
||||
gomock.Any(), gomock.Eq(
|
||||
&protos.BestOptionsRequest{
|
||||
Options: []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
|
||||
NodeInfoMap: grpcNodeInfoMap,
|
||||
Options: []*protos.Option{&grpcEoT2Micro, &grpcEoT2Large, &grpcEoT3Large, &grpcEoM44XLarge},
|
||||
NodeMap: grpcNodeInfoMap,
|
||||
})).Return(&tc.mockResponse, tc.errResponse)
|
||||
resp := g.BestOptions(options, tc.nodeInfo)
|
||||
|
||||
|
|
|
|||
|
|
@ -43,7 +43,7 @@ type BestOptionsRequest struct {
|
|||
|
||||
Options []*Option `protobuf:"bytes,1,rep,name=options,proto3" json:"options,omitempty"`
|
||||
// key is node id from options
|
||||
NodeInfoMap map[string]*v1.Node `protobuf:"bytes,2,rep,name=nodeInfoMap,proto3" json:"nodeInfoMap,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
|
||||
NodeMap map[string]*v1.Node `protobuf:"bytes,2,rep,name=nodeMap,proto3" json:"nodeMap,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
|
||||
}
|
||||
|
||||
func (x *BestOptionsRequest) Reset() {
|
||||
|
|
@ -85,9 +85,9 @@ func (x *BestOptionsRequest) GetOptions() []*Option {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (x *BestOptionsRequest) GetNodeInfoMap() map[string]*v1.Node {
|
||||
func (x *BestOptionsRequest) GetNodeMap() map[string]*v1.Node {
|
||||
if x != nil {
|
||||
return x.NodeInfoMap
|
||||
return x.NodeMap
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
@ -220,18 +220,17 @@ var file_cluster_autoscaler_expander_grpcplugin_protos_expander_proto_rawDesc =
|
|||
0x65, 0x78, 0x70, 0x61, 0x6e, 0x64, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0a,
|
||||
0x67, 0x72, 0x70, 0x63, 0x70, 0x6c, 0x75, 0x67, 0x69, 0x6e, 0x1a, 0x22, 0x6b, 0x38, 0x73, 0x2e,
|
||||
0x69, 0x6f, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x63, 0x6f, 0x72, 0x65, 0x2f, 0x76, 0x31, 0x2f, 0x67,
|
||||
0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xef,
|
||||
0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0xdf,
|
||||
0x01, 0x0a, 0x12, 0x42, 0x65, 0x73, 0x74, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65,
|
||||
0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x2c, 0x0a, 0x07, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73,
|
||||
0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x70, 0x6c, 0x75,
|
||||
0x67, 0x69, 0x6e, 0x2e, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x07, 0x6f, 0x70, 0x74, 0x69,
|
||||
0x6f, 0x6e, 0x73, 0x12, 0x51, 0x0a, 0x0b, 0x6e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66, 0x6f, 0x4d,
|
||||
0x61, 0x70, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x2f, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x70,
|
||||
0x6c, 0x75, 0x67, 0x69, 0x6e, 0x2e, 0x42, 0x65, 0x73, 0x74, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e,
|
||||
0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x2e, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e, 0x66,
|
||||
0x6f, 0x4d, 0x61, 0x70, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0b, 0x6e, 0x6f, 0x64, 0x65, 0x49,
|
||||
0x6e, 0x66, 0x6f, 0x4d, 0x61, 0x70, 0x1a, 0x58, 0x0a, 0x10, 0x4e, 0x6f, 0x64, 0x65, 0x49, 0x6e,
|
||||
0x66, 0x6f, 0x4d, 0x61, 0x70, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65,
|
||||
0x6f, 0x6e, 0x73, 0x12, 0x45, 0x0a, 0x07, 0x6e, 0x6f, 0x64, 0x65, 0x4d, 0x61, 0x70, 0x18, 0x02,
|
||||
0x20, 0x03, 0x28, 0x0b, 0x32, 0x2b, 0x2e, 0x67, 0x72, 0x70, 0x63, 0x70, 0x6c, 0x75, 0x67, 0x69,
|
||||
0x6e, 0x2e, 0x42, 0x65, 0x73, 0x74, 0x4f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x52, 0x65, 0x71,
|
||||
0x75, 0x65, 0x73, 0x74, 0x2e, 0x4e, 0x6f, 0x64, 0x65, 0x4d, 0x61, 0x70, 0x45, 0x6e, 0x74, 0x72,
|
||||
0x79, 0x52, 0x07, 0x6e, 0x6f, 0x64, 0x65, 0x4d, 0x61, 0x70, 0x1a, 0x54, 0x0a, 0x0c, 0x4e, 0x6f,
|
||||
0x64, 0x65, 0x4d, 0x61, 0x70, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65,
|
||||
0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x2e, 0x0a, 0x05,
|
||||
0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x18, 0x2e, 0x6b, 0x38,
|
||||
0x73, 0x2e, 0x69, 0x6f, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x63, 0x6f, 0x72, 0x65, 0x2e, 0x76, 0x31,
|
||||
|
|
@ -278,16 +277,16 @@ var file_cluster_autoscaler_expander_grpcplugin_protos_expander_proto_goTypes =
|
|||
(*BestOptionsRequest)(nil), // 0: grpcplugin.BestOptionsRequest
|
||||
(*BestOptionsResponse)(nil), // 1: grpcplugin.BestOptionsResponse
|
||||
(*Option)(nil), // 2: grpcplugin.Option
|
||||
nil, // 3: grpcplugin.BestOptionsRequest.NodeInfoMapEntry
|
||||
nil, // 3: grpcplugin.BestOptionsRequest.NodeMapEntry
|
||||
(*v1.Pod)(nil), // 4: k8s.io.api.core.v1.Pod
|
||||
(*v1.Node)(nil), // 5: k8s.io.api.core.v1.Node
|
||||
}
|
||||
var file_cluster_autoscaler_expander_grpcplugin_protos_expander_proto_depIdxs = []int32{
|
||||
2, // 0: grpcplugin.BestOptionsRequest.options:type_name -> grpcplugin.Option
|
||||
3, // 1: grpcplugin.BestOptionsRequest.nodeInfoMap:type_name -> grpcplugin.BestOptionsRequest.NodeInfoMapEntry
|
||||
3, // 1: grpcplugin.BestOptionsRequest.nodeMap:type_name -> grpcplugin.BestOptionsRequest.NodeMapEntry
|
||||
2, // 2: grpcplugin.BestOptionsResponse.options:type_name -> grpcplugin.Option
|
||||
4, // 3: grpcplugin.Option.pod:type_name -> k8s.io.api.core.v1.Pod
|
||||
5, // 4: grpcplugin.BestOptionsRequest.NodeInfoMapEntry.value:type_name -> k8s.io.api.core.v1.Node
|
||||
5, // 4: grpcplugin.BestOptionsRequest.NodeMapEntry.value:type_name -> k8s.io.api.core.v1.Node
|
||||
0, // 5: grpcplugin.Expander.BestOptions:input_type -> grpcplugin.BestOptionsRequest
|
||||
1, // 6: grpcplugin.Expander.BestOptions:output_type -> grpcplugin.BestOptionsResponse
|
||||
6, // [6:7] is the sub-list for method output_type
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ syntax = "proto3";
|
|||
|
||||
package grpcplugin;
|
||||
import "k8s.io/api/core/v1/generated.proto";
|
||||
//import "google/protobuf/struct.proto";
|
||||
option go_package = "cluster-autoscaler/expander/grpcplugin/protos";
|
||||
|
||||
|
||||
|
|
@ -17,7 +16,7 @@ service Expander {
|
|||
message BestOptionsRequest {
|
||||
repeated Option options = 1;
|
||||
// key is node id from options
|
||||
map<string, k8s.io.api.core.v1.Node> nodeInfoMap = 2;
|
||||
map<string, k8s.io.api.core.v1.Node> nodeMap = 2;
|
||||
}
|
||||
message BestOptionsResponse {
|
||||
repeated Option options = 1;
|
||||
|
|
|
|||
|
|
@ -791,8 +791,6 @@ golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8U
|
|||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.0.0-20201002170205-7f63de1d35b0/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
|
||||
golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a h1:kr2P4QFmQr29mSLA43kwrOcgcReGTfbE9N577tCTuBc=
|
||||
golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8=
|
||||
golang.org/x/crypto v0.0.0-20210817164053-32db794688a5/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
|
||||
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 h1:7I4JAnoQBe7ZtJcBaYHi5UtiO8tQHbUSXxL+pnGRANg=
|
||||
|
|
|
|||
|
|
@ -71,7 +71,7 @@ func (intf *Interface) addImports(im map[string]bool) {
|
|||
}
|
||||
}
|
||||
|
||||
// AddMethod adds a new method, deduplicating by method name.
|
||||
// AddMethod adds a new method, de-duplicating by method name.
|
||||
func (intf *Interface) AddMethod(m *Method) {
|
||||
for _, me := range intf.Methods {
|
||||
if me.Name == m.Name {
|
||||
|
|
@ -260,11 +260,10 @@ func (mt *MapType) addImports(im map[string]bool) {
|
|||
// NamedType is an exported type in a package.
|
||||
type NamedType struct {
|
||||
Package string // may be empty
|
||||
Type string // TODO: should this be typed Type?
|
||||
Type string
|
||||
}
|
||||
|
||||
func (nt *NamedType) String(pm map[string]string, pkgOverride string) string {
|
||||
// TODO: is this right?
|
||||
if pkgOverride == nt.Package {
|
||||
return nt.Type
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,445 +0,0 @@
|
|||
/*
|
||||
Copyright 2014 The Kubernetes 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 clock
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// PassiveClock allows for injecting fake or real clocks into code
|
||||
// that needs to read the current time but does not support scheduling
|
||||
// activity in the future.
|
||||
type PassiveClock interface {
|
||||
Now() time.Time
|
||||
Since(time.Time) time.Duration
|
||||
}
|
||||
|
||||
// Clock allows for injecting fake or real clocks into code that
|
||||
// needs to do arbitrary things based on time.
|
||||
type Clock interface {
|
||||
PassiveClock
|
||||
After(time.Duration) <-chan time.Time
|
||||
AfterFunc(time.Duration, func()) Timer
|
||||
NewTimer(time.Duration) Timer
|
||||
Sleep(time.Duration)
|
||||
NewTicker(time.Duration) Ticker
|
||||
}
|
||||
|
||||
// RealClock really calls time.Now()
|
||||
type RealClock struct{}
|
||||
|
||||
// Now returns the current time.
|
||||
func (RealClock) Now() time.Time {
|
||||
return time.Now()
|
||||
}
|
||||
|
||||
// Since returns time since the specified timestamp.
|
||||
func (RealClock) Since(ts time.Time) time.Duration {
|
||||
return time.Since(ts)
|
||||
}
|
||||
|
||||
// After is the same as time.After(d).
|
||||
func (RealClock) After(d time.Duration) <-chan time.Time {
|
||||
return time.After(d)
|
||||
}
|
||||
|
||||
// AfterFunc is the same as time.AfterFunc(d, f).
|
||||
func (RealClock) AfterFunc(d time.Duration, f func()) Timer {
|
||||
return &realTimer{
|
||||
timer: time.AfterFunc(d, f),
|
||||
}
|
||||
}
|
||||
|
||||
// NewTimer returns a new Timer.
|
||||
func (RealClock) NewTimer(d time.Duration) Timer {
|
||||
return &realTimer{
|
||||
timer: time.NewTimer(d),
|
||||
}
|
||||
}
|
||||
|
||||
// NewTicker returns a new Ticker.
|
||||
func (RealClock) NewTicker(d time.Duration) Ticker {
|
||||
return &realTicker{
|
||||
ticker: time.NewTicker(d),
|
||||
}
|
||||
}
|
||||
|
||||
// Sleep pauses the RealClock for duration d.
|
||||
func (RealClock) Sleep(d time.Duration) {
|
||||
time.Sleep(d)
|
||||
}
|
||||
|
||||
// FakePassiveClock implements PassiveClock, but returns an arbitrary time.
|
||||
type FakePassiveClock struct {
|
||||
lock sync.RWMutex
|
||||
time time.Time
|
||||
}
|
||||
|
||||
// FakeClock implements Clock, but returns an arbitrary time.
|
||||
type FakeClock struct {
|
||||
FakePassiveClock
|
||||
|
||||
// waiters are waiting for the fake time to pass their specified time
|
||||
waiters []fakeClockWaiter
|
||||
}
|
||||
|
||||
type fakeClockWaiter struct {
|
||||
targetTime time.Time
|
||||
stepInterval time.Duration
|
||||
skipIfBlocked bool
|
||||
destChan chan time.Time
|
||||
afterFunc func()
|
||||
}
|
||||
|
||||
// NewFakePassiveClock returns a new FakePassiveClock.
|
||||
func NewFakePassiveClock(t time.Time) *FakePassiveClock {
|
||||
return &FakePassiveClock{
|
||||
time: t,
|
||||
}
|
||||
}
|
||||
|
||||
// NewFakeClock returns a new FakeClock
|
||||
func NewFakeClock(t time.Time) *FakeClock {
|
||||
return &FakeClock{
|
||||
FakePassiveClock: *NewFakePassiveClock(t),
|
||||
}
|
||||
}
|
||||
|
||||
// Now returns f's time.
|
||||
func (f *FakePassiveClock) Now() time.Time {
|
||||
f.lock.RLock()
|
||||
defer f.lock.RUnlock()
|
||||
return f.time
|
||||
}
|
||||
|
||||
// Since returns time since the time in f.
|
||||
func (f *FakePassiveClock) Since(ts time.Time) time.Duration {
|
||||
f.lock.RLock()
|
||||
defer f.lock.RUnlock()
|
||||
return f.time.Sub(ts)
|
||||
}
|
||||
|
||||
// SetTime sets the time on the FakePassiveClock.
|
||||
func (f *FakePassiveClock) SetTime(t time.Time) {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
f.time = t
|
||||
}
|
||||
|
||||
// After is the Fake version of time.After(d).
|
||||
func (f *FakeClock) After(d time.Duration) <-chan time.Time {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
stopTime := f.time.Add(d)
|
||||
ch := make(chan time.Time, 1) // Don't block!
|
||||
f.waiters = append(f.waiters, fakeClockWaiter{
|
||||
targetTime: stopTime,
|
||||
destChan: ch,
|
||||
})
|
||||
return ch
|
||||
}
|
||||
|
||||
// AfterFunc is the Fake version of time.AfterFunc(d, callback).
|
||||
func (f *FakeClock) AfterFunc(d time.Duration, cb func()) Timer {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
stopTime := f.time.Add(d)
|
||||
ch := make(chan time.Time, 1) // Don't block!
|
||||
|
||||
timer := &fakeTimer{
|
||||
fakeClock: f,
|
||||
waiter: fakeClockWaiter{
|
||||
targetTime: stopTime,
|
||||
destChan: ch,
|
||||
afterFunc: cb,
|
||||
},
|
||||
}
|
||||
f.waiters = append(f.waiters, timer.waiter)
|
||||
return timer
|
||||
}
|
||||
|
||||
// NewTimer is the Fake version of time.NewTimer(d).
|
||||
func (f *FakeClock) NewTimer(d time.Duration) Timer {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
stopTime := f.time.Add(d)
|
||||
ch := make(chan time.Time, 1) // Don't block!
|
||||
timer := &fakeTimer{
|
||||
fakeClock: f,
|
||||
waiter: fakeClockWaiter{
|
||||
targetTime: stopTime,
|
||||
destChan: ch,
|
||||
},
|
||||
}
|
||||
f.waiters = append(f.waiters, timer.waiter)
|
||||
return timer
|
||||
}
|
||||
|
||||
// NewTicker returns a new Ticker.
|
||||
func (f *FakeClock) NewTicker(d time.Duration) Ticker {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
tickTime := f.time.Add(d)
|
||||
ch := make(chan time.Time, 1) // hold one tick
|
||||
f.waiters = append(f.waiters, fakeClockWaiter{
|
||||
targetTime: tickTime,
|
||||
stepInterval: d,
|
||||
skipIfBlocked: true,
|
||||
destChan: ch,
|
||||
})
|
||||
|
||||
return &fakeTicker{
|
||||
c: ch,
|
||||
}
|
||||
}
|
||||
|
||||
// Step moves clock by Duration, notifies anyone that's called After, Tick, or NewTimer
|
||||
func (f *FakeClock) Step(d time.Duration) {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
f.setTimeLocked(f.time.Add(d))
|
||||
}
|
||||
|
||||
// SetTime sets the time on a FakeClock.
|
||||
func (f *FakeClock) SetTime(t time.Time) {
|
||||
f.lock.Lock()
|
||||
defer f.lock.Unlock()
|
||||
f.setTimeLocked(t)
|
||||
}
|
||||
|
||||
// Actually changes the time and checks any waiters. f must be write-locked.
|
||||
func (f *FakeClock) setTimeLocked(t time.Time) {
|
||||
f.time = t
|
||||
newWaiters := make([]fakeClockWaiter, 0, len(f.waiters))
|
||||
for i := range f.waiters {
|
||||
w := &f.waiters[i]
|
||||
if !w.targetTime.After(t) {
|
||||
|
||||
if w.skipIfBlocked {
|
||||
select {
|
||||
case w.destChan <- t:
|
||||
default:
|
||||
}
|
||||
} else {
|
||||
w.destChan <- t
|
||||
}
|
||||
|
||||
if w.afterFunc != nil {
|
||||
w.afterFunc()
|
||||
}
|
||||
|
||||
if w.stepInterval > 0 {
|
||||
for !w.targetTime.After(t) {
|
||||
w.targetTime = w.targetTime.Add(w.stepInterval)
|
||||
}
|
||||
newWaiters = append(newWaiters, *w)
|
||||
}
|
||||
|
||||
} else {
|
||||
newWaiters = append(newWaiters, f.waiters[i])
|
||||
}
|
||||
}
|
||||
f.waiters = newWaiters
|
||||
}
|
||||
|
||||
// HasWaiters returns true if After or AfterFunc has been called on f but not yet satisfied
|
||||
// (so you can write race-free tests).
|
||||
func (f *FakeClock) HasWaiters() bool {
|
||||
f.lock.RLock()
|
||||
defer f.lock.RUnlock()
|
||||
return len(f.waiters) > 0
|
||||
}
|
||||
|
||||
// Sleep pauses the FakeClock for duration d.
|
||||
func (f *FakeClock) Sleep(d time.Duration) {
|
||||
f.Step(d)
|
||||
}
|
||||
|
||||
// IntervalClock implements Clock, but each invocation of Now steps the clock forward the specified duration
|
||||
type IntervalClock struct {
|
||||
Time time.Time
|
||||
Duration time.Duration
|
||||
}
|
||||
|
||||
// Now returns i's time.
|
||||
func (i *IntervalClock) Now() time.Time {
|
||||
i.Time = i.Time.Add(i.Duration)
|
||||
return i.Time
|
||||
}
|
||||
|
||||
// Since returns time since the time in i.
|
||||
func (i *IntervalClock) Since(ts time.Time) time.Duration {
|
||||
return i.Time.Sub(ts)
|
||||
}
|
||||
|
||||
// After is currently unimplemented, will panic.
|
||||
// TODO: make interval clock use FakeClock so this can be implemented.
|
||||
func (*IntervalClock) After(d time.Duration) <-chan time.Time {
|
||||
panic("IntervalClock doesn't implement After")
|
||||
}
|
||||
|
||||
// AfterFunc is currently unimplemented, will panic.
|
||||
// TODO: make interval clock use FakeClock so this can be implemented.
|
||||
func (*IntervalClock) AfterFunc(d time.Duration, cb func()) Timer {
|
||||
panic("IntervalClock doesn't implement AfterFunc")
|
||||
}
|
||||
|
||||
// NewTimer is currently unimplemented, will panic.
|
||||
// TODO: make interval clock use FakeClock so this can be implemented.
|
||||
func (*IntervalClock) NewTimer(d time.Duration) Timer {
|
||||
panic("IntervalClock doesn't implement NewTimer")
|
||||
}
|
||||
|
||||
// NewTicker is currently unimplemented, will panic.
|
||||
// TODO: make interval clock use FakeClock so this can be implemented.
|
||||
func (*IntervalClock) NewTicker(d time.Duration) Ticker {
|
||||
panic("IntervalClock doesn't implement NewTicker")
|
||||
}
|
||||
|
||||
// Sleep is currently unimplemented; will panic.
|
||||
func (*IntervalClock) Sleep(d time.Duration) {
|
||||
panic("IntervalClock doesn't implement Sleep")
|
||||
}
|
||||
|
||||
// Timer allows for injecting fake or real timers into code that
|
||||
// needs to do arbitrary things based on time.
|
||||
type Timer interface {
|
||||
C() <-chan time.Time
|
||||
Stop() bool
|
||||
Reset(d time.Duration) bool
|
||||
}
|
||||
|
||||
// realTimer is backed by an actual time.Timer.
|
||||
type realTimer struct {
|
||||
timer *time.Timer
|
||||
}
|
||||
|
||||
// C returns the underlying timer's channel.
|
||||
func (r *realTimer) C() <-chan time.Time {
|
||||
return r.timer.C
|
||||
}
|
||||
|
||||
// Stop calls Stop() on the underlying timer.
|
||||
func (r *realTimer) Stop() bool {
|
||||
return r.timer.Stop()
|
||||
}
|
||||
|
||||
// Reset calls Reset() on the underlying timer.
|
||||
func (r *realTimer) Reset(d time.Duration) bool {
|
||||
return r.timer.Reset(d)
|
||||
}
|
||||
|
||||
// fakeTimer implements Timer based on a FakeClock.
|
||||
type fakeTimer struct {
|
||||
fakeClock *FakeClock
|
||||
waiter fakeClockWaiter
|
||||
}
|
||||
|
||||
// C returns the channel that notifies when this timer has fired.
|
||||
func (f *fakeTimer) C() <-chan time.Time {
|
||||
return f.waiter.destChan
|
||||
}
|
||||
|
||||
// Stop conditionally stops the timer. If the timer has neither fired
|
||||
// nor been stopped then this call stops the timer and returns true,
|
||||
// otherwise this call returns false. This is like time.Timer::Stop.
|
||||
func (f *fakeTimer) Stop() bool {
|
||||
f.fakeClock.lock.Lock()
|
||||
defer f.fakeClock.lock.Unlock()
|
||||
// The timer has already fired or been stopped, unless it is found
|
||||
// among the clock's waiters.
|
||||
stopped := false
|
||||
oldWaiters := f.fakeClock.waiters
|
||||
newWaiters := make([]fakeClockWaiter, 0, len(oldWaiters))
|
||||
seekChan := f.waiter.destChan
|
||||
for i := range oldWaiters {
|
||||
// Identify the timer's fakeClockWaiter by the identity of the
|
||||
// destination channel, nothing else is necessarily unique and
|
||||
// constant since the timer's creation.
|
||||
if oldWaiters[i].destChan == seekChan {
|
||||
stopped = true
|
||||
} else {
|
||||
newWaiters = append(newWaiters, oldWaiters[i])
|
||||
}
|
||||
}
|
||||
|
||||
f.fakeClock.waiters = newWaiters
|
||||
|
||||
return stopped
|
||||
}
|
||||
|
||||
// Reset conditionally updates the firing time of the timer. If the
|
||||
// timer has neither fired nor been stopped then this call resets the
|
||||
// timer to the fake clock's "now" + d and returns true, otherwise
|
||||
// it creates a new waiter, adds it to the clock, and returns true.
|
||||
//
|
||||
// It is not possible to return false, because a fake timer can be reset
|
||||
// from any state (waiting to fire, already fired, and stopped).
|
||||
//
|
||||
// See the GoDoc for time.Timer::Reset for more context on why
|
||||
// the return value of Reset() is not useful.
|
||||
func (f *fakeTimer) Reset(d time.Duration) bool {
|
||||
f.fakeClock.lock.Lock()
|
||||
defer f.fakeClock.lock.Unlock()
|
||||
waiters := f.fakeClock.waiters
|
||||
seekChan := f.waiter.destChan
|
||||
for i := range waiters {
|
||||
if waiters[i].destChan == seekChan {
|
||||
waiters[i].targetTime = f.fakeClock.time.Add(d)
|
||||
return true
|
||||
}
|
||||
}
|
||||
// No existing waiter, timer has already fired or been reset.
|
||||
// We should still enable Reset() to succeed by creating a
|
||||
// new waiter and adding it to the clock's waiters.
|
||||
newWaiter := fakeClockWaiter{
|
||||
targetTime: f.fakeClock.time.Add(d),
|
||||
destChan: seekChan,
|
||||
}
|
||||
f.fakeClock.waiters = append(f.fakeClock.waiters, newWaiter)
|
||||
return true
|
||||
}
|
||||
|
||||
// Ticker defines the Ticker interface
|
||||
type Ticker interface {
|
||||
C() <-chan time.Time
|
||||
Stop()
|
||||
}
|
||||
|
||||
type realTicker struct {
|
||||
ticker *time.Ticker
|
||||
}
|
||||
|
||||
func (t *realTicker) C() <-chan time.Time {
|
||||
return t.ticker.C
|
||||
}
|
||||
|
||||
func (t *realTicker) Stop() {
|
||||
t.ticker.Stop()
|
||||
}
|
||||
|
||||
type fakeTicker struct {
|
||||
c <-chan time.Time
|
||||
}
|
||||
|
||||
func (t *fakeTicker) C() <-chan time.Time {
|
||||
return t.c
|
||||
}
|
||||
|
||||
func (t *fakeTicker) Stop() {
|
||||
}
|
||||
|
|
@ -791,6 +791,7 @@ google.golang.org/genproto/googleapis/api/httpbody
|
|||
google.golang.org/genproto/googleapis/rpc/status
|
||||
google.golang.org/genproto/protobuf/field_mask
|
||||
# google.golang.org/grpc v1.40.0
|
||||
## explicit
|
||||
google.golang.org/grpc
|
||||
google.golang.org/grpc/attributes
|
||||
google.golang.org/grpc/backoff
|
||||
|
|
@ -839,6 +840,7 @@ google.golang.org/grpc/stats
|
|||
google.golang.org/grpc/status
|
||||
google.golang.org/grpc/tap
|
||||
# google.golang.org/protobuf v1.27.1
|
||||
## explicit
|
||||
google.golang.org/protobuf/encoding/protojson
|
||||
google.golang.org/protobuf/encoding/prototext
|
||||
google.golang.org/protobuf/encoding/protowire
|
||||
|
|
@ -970,7 +972,6 @@ k8s.io/apimachinery/pkg/runtime/serializer/versioning
|
|||
k8s.io/apimachinery/pkg/selection
|
||||
k8s.io/apimachinery/pkg/types
|
||||
k8s.io/apimachinery/pkg/util/cache
|
||||
k8s.io/apimachinery/pkg/util/clock
|
||||
k8s.io/apimachinery/pkg/util/diff
|
||||
k8s.io/apimachinery/pkg/util/errors
|
||||
k8s.io/apimachinery/pkg/util/framer
|
||||
|
|
|
|||
Loading…
Reference in New Issue