diff --git a/hack/.golint_failures b/hack/.golint_failures index 2b2798e46a7..0853347106b 100644 --- a/hack/.golint_failures +++ b/hack/.golint_failures @@ -166,6 +166,7 @@ pkg/kubelet/apis/deviceplugin/v1alpha pkg/kubelet/apis/deviceplugin/v1beta1 pkg/kubelet/apis/kubeletconfig pkg/kubelet/apis/kubeletconfig/v1beta1 +pkg/kubelet/apis/pluginregistration/v1alpha1 pkg/kubelet/cadvisor pkg/kubelet/cadvisor/testing pkg/kubelet/checkpoint @@ -217,6 +218,9 @@ pkg/kubelet/sysctl pkg/kubelet/types pkg/kubelet/util pkg/kubelet/util/cache +pkg/kubelet/util/pluginwatcher +pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1 +pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2 pkg/kubelet/util/queue pkg/kubelet/util/sliceutils pkg/kubemark diff --git a/hack/update-generated-kubelet-plugin-registration-dockerized.sh b/hack/update-generated-kubelet-plugin-registration-dockerized.sh new file mode 100755 index 00000000000..daf5abbd36a --- /dev/null +++ b/hack/update-generated-kubelet-plugin-registration-dockerized.sh @@ -0,0 +1,29 @@ +#!/bin/bash + +# Copyright 2018 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. + +set -o errexit +set -o nounset +set -o pipefail + +KUBE_ROOT="$(cd "$(dirname "${BASH_SOURCE}")/../" && pwd -P)" +KUBELET_PLUGIN_REGISTRATION_ROOT="${KUBE_ROOT}/pkg/kubelet/apis/pluginregistration/v1alpha1/" +KUBELET_EXAMPLE_PLUGIN_V1BETA1="${KUBE_ROOT}/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/" +KUBELET_EXAMPLE_PLUGIN_V1BETA2="${KUBE_ROOT}/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/" + +source "${KUBE_ROOT}/hack/lib/protoc.sh" +kube::protoc::generate_proto ${KUBELET_PLUGIN_REGISTRATION_ROOT} +kube::protoc::generate_proto ${KUBELET_EXAMPLE_PLUGIN_V1BETA1} +kube::protoc::generate_proto ${KUBELET_EXAMPLE_PLUGIN_V1BETA2} diff --git a/hack/update-generated-kubelet-plugin-registration.sh b/hack/update-generated-kubelet-plugin-registration.sh new file mode 100755 index 00000000000..308733c0246 --- /dev/null +++ b/hack/update-generated-kubelet-plugin-registration.sh @@ -0,0 +1,27 @@ +#!/bin/bash + +# Copyright 2018 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. + +set -o errexit +set -o nounset +set -o pipefail + +KUBE_ROOT=$(dirname "${BASH_SOURCE}")/.. + +# NOTE: All output from this script needs to be copied back to the calling +# source tree. This is managed in kube::build::copy_output in build/common.sh. +# If the output set is changed update that function. + +${KUBE_ROOT}/build/run.sh hack/update-generated-kubelet-plugin-registration-dockerized.sh "$@" diff --git a/hack/verify-generated-kubelet-plugin-registration.sh b/hack/verify-generated-kubelet-plugin-registration.sh new file mode 100755 index 00000000000..3dfffa8dcc0 --- /dev/null +++ b/hack/verify-generated-kubelet-plugin-registration.sh @@ -0,0 +1,39 @@ +#!/bin/bash + +# Copyright 2018 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. + +set -o errexit +set -o nounset +set -o pipefail + +KUBE_ROOT=$(dirname "${BASH_SOURCE}")/.. +ERROR="Kubelet Plugin Registration api is out of date. Please run hack/update-generated-kubelet-plugin-registration.sh" +KUBELET_PLUGIN_REGISTRATION_ROOT="${KUBE_ROOT}/pkg/kubelet/apis/pluginregistration/v1alpha1/" + +source "${KUBE_ROOT}/hack/lib/protoc.sh" +kube::golang::setup_env + +function cleanup { + rm -rf ${KUBELET_PLUGIN_REGISTRATION_ROOT}/_tmp/ +} + +trap cleanup EXIT + +mkdir -p ${KUBELET_PLUGIN_REGISTRATION_ROOT}/_tmp +cp ${KUBELET_PLUGIN_REGISTRATION_ROOT}/api.pb.go ${KUBELET_PLUGIN_REGISTRATION_ROOT}/_tmp/ + +KUBE_VERBOSE=3 "${KUBE_ROOT}/hack/update-generated-kubelet-plugin-registration.sh" +kube::protoc::diff "${KUBELET_PLUGIN_REGISTRATION_ROOT}/api.pb.go" "${KUBELET_PLUGIN_REGISTRATION_ROOT}/_tmp/api.pb.go" ${ERROR} +echo "Generated Kubelet Plugin Registration api is up to date." diff --git a/pkg/kubelet/BUILD b/pkg/kubelet/BUILD index b4d6d166fa9..710af32b7a3 100644 --- a/pkg/kubelet/BUILD +++ b/pkg/kubelet/BUILD @@ -84,6 +84,7 @@ go_library( "//pkg/kubelet/util:go_default_library", "//pkg/kubelet/util/format:go_default_library", "//pkg/kubelet/util/manager:go_default_library", + "//pkg/kubelet/util/pluginwatcher:go_default_library", "//pkg/kubelet/util/queue:go_default_library", "//pkg/kubelet/util/sliceutils:go_default_library", "//pkg/kubelet/volumemanager:go_default_library", diff --git a/pkg/kubelet/apis/BUILD b/pkg/kubelet/apis/BUILD index 47cb8184ccb..2a22e48121a 100644 --- a/pkg/kubelet/apis/BUILD +++ b/pkg/kubelet/apis/BUILD @@ -41,6 +41,7 @@ filegroup( "//pkg/kubelet/apis/deviceplugin/v1alpha:all-srcs", "//pkg/kubelet/apis/deviceplugin/v1beta1:all-srcs", "//pkg/kubelet/apis/kubeletconfig:all-srcs", + "//pkg/kubelet/apis/pluginregistration/v1alpha1:all-srcs", "//pkg/kubelet/apis/stats/v1alpha1:all-srcs", ], tags = ["automanaged"], diff --git a/pkg/kubelet/apis/pluginregistration/v1alpha1/BUILD b/pkg/kubelet/apis/pluginregistration/v1alpha1/BUILD new file mode 100644 index 00000000000..f51668500bb --- /dev/null +++ b/pkg/kubelet/apis/pluginregistration/v1alpha1/BUILD @@ -0,0 +1,40 @@ +package(default_visibility = ["//visibility:public"]) + +load( + "@io_bazel_rules_go//go:def.bzl", + "go_library", +) + +go_library( + name = "go_default_library", + srcs = [ + "api.pb.go", + "constants.go", + ], + importpath = "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1alpha1", + deps = [ + "//vendor/github.com/gogo/protobuf/gogoproto:go_default_library", + "//vendor/github.com/gogo/protobuf/proto:go_default_library", + "//vendor/golang.org/x/net/context:go_default_library", + "//vendor/google.golang.org/grpc:go_default_library", + ], +) + +filegroup( + name = "package-srcs", + srcs = glob(["**"]), + tags = ["automanaged"], + visibility = ["//visibility:private"], +) + +filegroup( + name = "all-srcs", + srcs = [":package-srcs"], + tags = ["automanaged"], +) + +filegroup( + name = "go_default_library_protos", + srcs = ["api.proto"], + visibility = ["//visibility:public"], +) diff --git a/pkg/kubelet/apis/pluginregistration/v1alpha1/api.pb.go b/pkg/kubelet/apis/pluginregistration/v1alpha1/api.pb.go new file mode 100644 index 00000000000..96e1d571dba --- /dev/null +++ b/pkg/kubelet/apis/pluginregistration/v1alpha1/api.pb.go @@ -0,0 +1,1027 @@ +/* +Copyright 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. +*/ + +// Code generated by protoc-gen-gogo. +// source: api.proto +// DO NOT EDIT! + +/* + Package pluginregistration is a generated protocol buffer package. + + It is generated from these files: + api.proto + + It has these top-level messages: + PluginInfo + RegistrationStatus + RegistrationStatusResponse + InfoRequest +*/ +package pluginregistration + +import proto "github.com/gogo/protobuf/proto" +import fmt "fmt" +import math "math" +import _ "github.com/gogo/protobuf/gogoproto" + +import ( + context "golang.org/x/net/context" + grpc "google.golang.org/grpc" +) + +import strings "strings" +import reflect "reflect" + +import io "io" + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package + +// PluginInfo is the message sent from a plugin to the Kubelet pluginwatcher for plugin registration +type PluginInfo struct { + // Type of the Plugin. CSIPlugin or DevicePlugin + Type string `protobuf:"bytes,1,opt,name=type,proto3" json:"type,omitempty"` + // Plugin name that uniquely identifies the plugin for the given plugin type. + // For DevicePlugin, this is the resource name that the plugin manages and + // should follow the extended resource name convention. + // For CSI, this is the CSI driver registrar name. + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` + // Optional endpoint location. If found set by Kubelet component, + // Kubelet component will use this endpoint for specific requests. + // This allows the plugin to register using one endpoint and possibly use + // a different socket for control operations. CSI uses this model to delegate + // its registration external from the plugin. + Endpoint string `protobuf:"bytes,3,opt,name=endpoint,proto3" json:"endpoint,omitempty"` + // Plugin service API versions the plugin supports. + // For DevicePlugin, this maps to the deviceplugin API versions the + // plugin supports at the given socket. + // The Kubelet component communicating with the plugin should be able + // to choose any preferred version from this list, or returns an error + // if none of the listed versions is supported. + SupportedVersions []string `protobuf:"bytes,4,rep,name=supported_versions,json=supportedVersions" json:"supported_versions,omitempty"` +} + +func (m *PluginInfo) Reset() { *m = PluginInfo{} } +func (*PluginInfo) ProtoMessage() {} +func (*PluginInfo) Descriptor() ([]byte, []int) { return fileDescriptorApi, []int{0} } + +func (m *PluginInfo) GetType() string { + if m != nil { + return m.Type + } + return "" +} + +func (m *PluginInfo) GetName() string { + if m != nil { + return m.Name + } + return "" +} + +func (m *PluginInfo) GetEndpoint() string { + if m != nil { + return m.Endpoint + } + return "" +} + +func (m *PluginInfo) GetSupportedVersions() []string { + if m != nil { + return m.SupportedVersions + } + return nil +} + +// RegistrationStatus is the message sent from Kubelet pluginwatcher to the plugin for notification on registration status +type RegistrationStatus struct { + // True if plugin gets registered successfully at Kubelet + PluginRegistered bool `protobuf:"varint,1,opt,name=plugin_registered,json=pluginRegistered,proto3" json:"plugin_registered,omitempty"` + // Error message in case plugin fails to register, empty string otherwise + Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` +} + +func (m *RegistrationStatus) Reset() { *m = RegistrationStatus{} } +func (*RegistrationStatus) ProtoMessage() {} +func (*RegistrationStatus) Descriptor() ([]byte, []int) { return fileDescriptorApi, []int{1} } + +func (m *RegistrationStatus) GetPluginRegistered() bool { + if m != nil { + return m.PluginRegistered + } + return false +} + +func (m *RegistrationStatus) GetError() string { + if m != nil { + return m.Error + } + return "" +} + +// RegistrationStatusResponse is sent by plugin to kubelet in response to RegistrationStatus RPC +type RegistrationStatusResponse struct { +} + +func (m *RegistrationStatusResponse) Reset() { *m = RegistrationStatusResponse{} } +func (*RegistrationStatusResponse) ProtoMessage() {} +func (*RegistrationStatusResponse) Descriptor() ([]byte, []int) { return fileDescriptorApi, []int{2} } + +// InfoRequest is the empty request message from Kubelet +type InfoRequest struct { +} + +func (m *InfoRequest) Reset() { *m = InfoRequest{} } +func (*InfoRequest) ProtoMessage() {} +func (*InfoRequest) Descriptor() ([]byte, []int) { return fileDescriptorApi, []int{3} } + +func init() { + proto.RegisterType((*PluginInfo)(nil), "pluginregistration.PluginInfo") + proto.RegisterType((*RegistrationStatus)(nil), "pluginregistration.RegistrationStatus") + proto.RegisterType((*RegistrationStatusResponse)(nil), "pluginregistration.RegistrationStatusResponse") + proto.RegisterType((*InfoRequest)(nil), "pluginregistration.InfoRequest") +} + +// Reference imports to suppress errors if they are not otherwise used. +var _ context.Context +var _ grpc.ClientConn + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +const _ = grpc.SupportPackageIsVersion4 + +// Client API for Registration service + +type RegistrationClient interface { + GetInfo(ctx context.Context, in *InfoRequest, opts ...grpc.CallOption) (*PluginInfo, error) + NotifyRegistrationStatus(ctx context.Context, in *RegistrationStatus, opts ...grpc.CallOption) (*RegistrationStatusResponse, error) +} + +type registrationClient struct { + cc *grpc.ClientConn +} + +func NewRegistrationClient(cc *grpc.ClientConn) RegistrationClient { + return ®istrationClient{cc} +} + +func (c *registrationClient) GetInfo(ctx context.Context, in *InfoRequest, opts ...grpc.CallOption) (*PluginInfo, error) { + out := new(PluginInfo) + err := grpc.Invoke(ctx, "/pluginregistration.Registration/GetInfo", in, out, c.cc, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *registrationClient) NotifyRegistrationStatus(ctx context.Context, in *RegistrationStatus, opts ...grpc.CallOption) (*RegistrationStatusResponse, error) { + out := new(RegistrationStatusResponse) + err := grpc.Invoke(ctx, "/pluginregistration.Registration/NotifyRegistrationStatus", in, out, c.cc, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// Server API for Registration service + +type RegistrationServer interface { + GetInfo(context.Context, *InfoRequest) (*PluginInfo, error) + NotifyRegistrationStatus(context.Context, *RegistrationStatus) (*RegistrationStatusResponse, error) +} + +func RegisterRegistrationServer(s *grpc.Server, srv RegistrationServer) { + s.RegisterService(&_Registration_serviceDesc, srv) +} + +func _Registration_GetInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(InfoRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(RegistrationServer).GetInfo(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/pluginregistration.Registration/GetInfo", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(RegistrationServer).GetInfo(ctx, req.(*InfoRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Registration_NotifyRegistrationStatus_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(RegistrationStatus) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(RegistrationServer).NotifyRegistrationStatus(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/pluginregistration.Registration/NotifyRegistrationStatus", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(RegistrationServer).NotifyRegistrationStatus(ctx, req.(*RegistrationStatus)) + } + return interceptor(ctx, in, info, handler) +} + +var _Registration_serviceDesc = grpc.ServiceDesc{ + ServiceName: "pluginregistration.Registration", + HandlerType: (*RegistrationServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "GetInfo", + Handler: _Registration_GetInfo_Handler, + }, + { + MethodName: "NotifyRegistrationStatus", + Handler: _Registration_NotifyRegistrationStatus_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "api.proto", +} + +func (m *PluginInfo) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *PluginInfo) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Type) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.Type))) + i += copy(dAtA[i:], m.Type) + } + if len(m.Name) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.Name))) + i += copy(dAtA[i:], m.Name) + } + if len(m.Endpoint) > 0 { + dAtA[i] = 0x1a + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.Endpoint))) + i += copy(dAtA[i:], m.Endpoint) + } + if len(m.SupportedVersions) > 0 { + for _, s := range m.SupportedVersions { + dAtA[i] = 0x22 + i++ + l = len(s) + for l >= 1<<7 { + dAtA[i] = uint8(uint64(l)&0x7f | 0x80) + l >>= 7 + i++ + } + dAtA[i] = uint8(l) + i++ + i += copy(dAtA[i:], s) + } + } + return i, nil +} + +func (m *RegistrationStatus) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *RegistrationStatus) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if m.PluginRegistered { + dAtA[i] = 0x8 + i++ + if m.PluginRegistered { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i++ + } + if len(m.Error) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.Error))) + i += copy(dAtA[i:], m.Error) + } + return i, nil +} + +func (m *RegistrationStatusResponse) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *RegistrationStatusResponse) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + return i, nil +} + +func (m *InfoRequest) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *InfoRequest) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + return i, nil +} + +func encodeFixed64Api(dAtA []byte, offset int, v uint64) int { + dAtA[offset] = uint8(v) + dAtA[offset+1] = uint8(v >> 8) + dAtA[offset+2] = uint8(v >> 16) + dAtA[offset+3] = uint8(v >> 24) + dAtA[offset+4] = uint8(v >> 32) + dAtA[offset+5] = uint8(v >> 40) + dAtA[offset+6] = uint8(v >> 48) + dAtA[offset+7] = uint8(v >> 56) + return offset + 8 +} +func encodeFixed32Api(dAtA []byte, offset int, v uint32) int { + dAtA[offset] = uint8(v) + dAtA[offset+1] = uint8(v >> 8) + dAtA[offset+2] = uint8(v >> 16) + dAtA[offset+3] = uint8(v >> 24) + return offset + 4 +} +func encodeVarintApi(dAtA []byte, offset int, v uint64) int { + for v >= 1<<7 { + dAtA[offset] = uint8(v&0x7f | 0x80) + v >>= 7 + offset++ + } + dAtA[offset] = uint8(v) + return offset + 1 +} +func (m *PluginInfo) Size() (n int) { + var l int + _ = l + l = len(m.Type) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + l = len(m.Name) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + l = len(m.Endpoint) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + if len(m.SupportedVersions) > 0 { + for _, s := range m.SupportedVersions { + l = len(s) + n += 1 + l + sovApi(uint64(l)) + } + } + return n +} + +func (m *RegistrationStatus) Size() (n int) { + var l int + _ = l + if m.PluginRegistered { + n += 2 + } + l = len(m.Error) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + return n +} + +func (m *RegistrationStatusResponse) Size() (n int) { + var l int + _ = l + return n +} + +func (m *InfoRequest) Size() (n int) { + var l int + _ = l + return n +} + +func sovApi(x uint64) (n int) { + for { + n++ + x >>= 7 + if x == 0 { + break + } + } + return n +} +func sozApi(x uint64) (n int) { + return sovApi(uint64((x << 1) ^ uint64((int64(x) >> 63)))) +} +func (this *PluginInfo) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&PluginInfo{`, + `Type:` + fmt.Sprintf("%v", this.Type) + `,`, + `Name:` + fmt.Sprintf("%v", this.Name) + `,`, + `Endpoint:` + fmt.Sprintf("%v", this.Endpoint) + `,`, + `SupportedVersions:` + fmt.Sprintf("%v", this.SupportedVersions) + `,`, + `}`, + }, "") + return s +} +func (this *RegistrationStatus) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&RegistrationStatus{`, + `PluginRegistered:` + fmt.Sprintf("%v", this.PluginRegistered) + `,`, + `Error:` + fmt.Sprintf("%v", this.Error) + `,`, + `}`, + }, "") + return s +} +func (this *RegistrationStatusResponse) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&RegistrationStatusResponse{`, + `}`, + }, "") + return s +} +func (this *InfoRequest) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&InfoRequest{`, + `}`, + }, "") + return s +} +func valueToStringApi(v interface{}) string { + rv := reflect.ValueOf(v) + if rv.IsNil() { + return "nil" + } + pv := reflect.Indirect(rv).Interface() + return fmt.Sprintf("*%v", pv) +} +func (m *PluginInfo) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: PluginInfo: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: PluginInfo: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Type", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Type = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Name", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Name = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 3: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Endpoint", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Endpoint = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 4: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field SupportedVersions", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.SupportedVersions = append(m.SupportedVersions, string(dAtA[iNdEx:postIndex])) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipApi(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthApi + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *RegistrationStatus) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: RegistrationStatus: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: RegistrationStatus: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field PluginRegistered", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + m.PluginRegistered = bool(v != 0) + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Error", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Error = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipApi(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthApi + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *RegistrationStatusResponse) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: RegistrationStatusResponse: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: RegistrationStatusResponse: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + default: + iNdEx = preIndex + skippy, err := skipApi(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthApi + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *InfoRequest) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: InfoRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: InfoRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + default: + iNdEx = preIndex + skippy, err := skipApi(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthApi + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func skipApi(dAtA []byte) (n int, err error) { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + wireType := int(wire & 0x7) + switch wireType { + case 0: + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + iNdEx++ + if dAtA[iNdEx-1] < 0x80 { + break + } + } + return iNdEx, nil + case 1: + iNdEx += 8 + return iNdEx, nil + case 2: + var length int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + length |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + iNdEx += length + if length < 0 { + return 0, ErrInvalidLengthApi + } + return iNdEx, nil + case 3: + for { + var innerWire uint64 + var start int = iNdEx + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + innerWire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + innerWireType := int(innerWire & 0x7) + if innerWireType == 4 { + break + } + next, err := skipApi(dAtA[start:]) + if err != nil { + return 0, err + } + iNdEx = start + next + } + return iNdEx, nil + case 4: + return iNdEx, nil + case 5: + iNdEx += 4 + return iNdEx, nil + default: + return 0, fmt.Errorf("proto: illegal wireType %d", wireType) + } + } + panic("unreachable") +} + +var ( + ErrInvalidLengthApi = fmt.Errorf("proto: negative length found during unmarshaling") + ErrIntOverflowApi = fmt.Errorf("proto: integer overflow") +) + +func init() { proto.RegisterFile("api.proto", fileDescriptorApi) } + +var fileDescriptorApi = []byte{ + // 337 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x52, 0x41, 0x4b, 0x33, 0x31, + 0x14, 0xdc, 0x7c, 0xed, 0xa7, 0xed, 0x53, 0xc1, 0x06, 0x0f, 0xcb, 0x52, 0x62, 0xd9, 0x83, 0x14, + 0xa4, 0x5b, 0xd0, 0x7f, 0xe0, 0x45, 0x04, 0x11, 0x89, 0xa0, 0xc7, 0xb2, 0xb5, 0xaf, 0x6b, 0xc0, + 0x26, 0x31, 0xc9, 0x0a, 0x3d, 0xe9, 0x4f, 0xf0, 0x67, 0xf5, 0x28, 0x9e, 0x3c, 0xda, 0xf5, 0x8f, + 0x48, 0xb3, 0x65, 0x2d, 0xb4, 0x07, 0x6f, 0x6f, 0xe6, 0x4d, 0x1e, 0x33, 0x43, 0xa0, 0x99, 0x6a, + 0x91, 0x68, 0xa3, 0x9c, 0xa2, 0x54, 0x3f, 0xe6, 0x99, 0x90, 0x06, 0x33, 0x61, 0x9d, 0x49, 0x9d, + 0x50, 0x32, 0xea, 0x65, 0xc2, 0x3d, 0xe4, 0xc3, 0xe4, 0x5e, 0x4d, 0xfa, 0x99, 0xca, 0x54, 0xdf, + 0x4b, 0x87, 0xf9, 0xd8, 0x23, 0x0f, 0xfc, 0x54, 0x9e, 0x88, 0x5f, 0x00, 0xae, 0xfd, 0x91, 0x0b, + 0x39, 0x56, 0x94, 0x42, 0xdd, 0x4d, 0x35, 0x86, 0xa4, 0x43, 0xba, 0x4d, 0xee, 0xe7, 0x05, 0x27, + 0xd3, 0x09, 0x86, 0xff, 0x4a, 0x6e, 0x31, 0xd3, 0x08, 0x1a, 0x28, 0x47, 0x5a, 0x09, 0xe9, 0xc2, + 0x9a, 0xe7, 0x2b, 0x4c, 0x7b, 0x40, 0x6d, 0xae, 0xb5, 0x32, 0x0e, 0x47, 0x83, 0x67, 0x34, 0x56, + 0x28, 0x69, 0xc3, 0x7a, 0xa7, 0xd6, 0x6d, 0xf2, 0x56, 0xb5, 0xb9, 0x5d, 0x2e, 0xe2, 0x3b, 0xa0, + 0x7c, 0xc5, 0xff, 0x8d, 0x4b, 0x5d, 0x6e, 0xe9, 0x31, 0xb4, 0xca, 0x6c, 0x83, 0x32, 0x1c, 0x1a, + 0x1c, 0x79, 0x57, 0x0d, 0xbe, 0x5f, 0x2e, 0x78, 0xc5, 0xd3, 0x03, 0xf8, 0x8f, 0xc6, 0x28, 0xb3, + 0xb4, 0x58, 0x82, 0xb8, 0x0d, 0xd1, 0xfa, 0x61, 0x8e, 0x56, 0x2b, 0x69, 0x31, 0xde, 0x83, 0x9d, + 0x45, 0x62, 0x8e, 0x4f, 0x39, 0x5a, 0x77, 0xf2, 0x41, 0x60, 0x77, 0x55, 0x4d, 0x2f, 0x61, 0xfb, + 0x1c, 0x9d, 0x2f, 0xe5, 0x30, 0x59, 0xaf, 0x39, 0x59, 0x79, 0x1c, 0xb1, 0x4d, 0x82, 0xdf, 0x56, + 0xe3, 0x80, 0x3a, 0x08, 0xaf, 0x94, 0x13, 0xe3, 0xe9, 0x86, 0xa8, 0x47, 0x9b, 0x5e, 0xaf, 0xeb, + 0xa2, 0xe4, 0x6f, 0xba, 0x2a, 0x61, 0x70, 0xd6, 0x9e, 0xcd, 0x19, 0xf9, 0x9c, 0xb3, 0xe0, 0xb5, + 0x60, 0x64, 0x56, 0x30, 0xf2, 0x5e, 0x30, 0xf2, 0x55, 0x30, 0xf2, 0xf6, 0xcd, 0x82, 0xe1, 0x96, + 0xff, 0x00, 0xa7, 0x3f, 0x01, 0x00, 0x00, 0xff, 0xff, 0xe4, 0xc0, 0xe3, 0x42, 0x50, 0x02, 0x00, + 0x00, +} diff --git a/pkg/kubelet/apis/pluginregistration/v1alpha1/api.proto b/pkg/kubelet/apis/pluginregistration/v1alpha1/api.proto new file mode 100644 index 00000000000..319b3f19fb7 --- /dev/null +++ b/pkg/kubelet/apis/pluginregistration/v1alpha1/api.proto @@ -0,0 +1,60 @@ +// To regenerate api.pb.go run hack/update-generated-kubelet-plugin-registration.sh +syntax = 'proto3'; + +package pluginregistration; + +import "github.com/gogo/protobuf/gogoproto/gogo.proto"; + +option (gogoproto.goproto_stringer_all) = false; +option (gogoproto.stringer_all) = true; +option (gogoproto.goproto_getters_all) = true; +option (gogoproto.marshaler_all) = true; +option (gogoproto.sizer_all) = true; +option (gogoproto.unmarshaler_all) = true; +option (gogoproto.goproto_unrecognized_all) = false; + +// PluginInfo is the message sent from a plugin to the Kubelet pluginwatcher for plugin registration +message PluginInfo { + // Type of the Plugin. CSIPlugin or DevicePlugin + string type = 1; + // Plugin name that uniquely identifies the plugin for the given plugin type. + // For DevicePlugin, this is the resource name that the plugin manages and + // should follow the extended resource name convention. + // For CSI, this is the CSI driver registrar name. + string name = 2; + // Optional endpoint location. If found set by Kubelet component, + // Kubelet component will use this endpoint for specific requests. + // This allows the plugin to register using one endpoint and possibly use + // a different socket for control operations. CSI uses this model to delegate + // its registration external from the plugin. + string endpoint = 3; + // Plugin service API versions the plugin supports. + // For DevicePlugin, this maps to the deviceplugin API versions the + // plugin supports at the given socket. + // The Kubelet component communicating with the plugin should be able + // to choose any preferred version from this list, or returns an error + // if none of the listed versions is supported. + repeated string supported_versions = 4; +} + +// RegistrationStatus is the message sent from Kubelet pluginwatcher to the plugin for notification on registration status +message RegistrationStatus { + // True if plugin gets registered successfully at Kubelet + bool plugin_registered = 1; + // Error message in case plugin fails to register, empty string otherwise + string error = 2; +} + +// RegistrationStatusResponse is sent by plugin to kubelet in response to RegistrationStatus RPC +message RegistrationStatusResponse { +} + +// InfoRequest is the empty request message from Kubelet +message InfoRequest { +} + +// Registration is the service advertised by the Plugins. +service Registration { + rpc GetInfo(InfoRequest) returns (PluginInfo) {} + rpc NotifyRegistrationStatus(RegistrationStatus) returns (RegistrationStatusResponse) {} +} diff --git a/pkg/kubelet/apis/pluginregistration/v1alpha1/constants.go b/pkg/kubelet/apis/pluginregistration/v1alpha1/constants.go new file mode 100644 index 00000000000..cfc1b7c6d7c --- /dev/null +++ b/pkg/kubelet/apis/pluginregistration/v1alpha1/constants.go @@ -0,0 +1,22 @@ +/* +Copyright 2018 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 pluginregistration + +const ( + CSIPlugin = "CSIPlugin" + DevicePlugin = "DevicePlugin" +) diff --git a/pkg/kubelet/kubelet.go b/pkg/kubelet/kubelet.go index 421c0c98bea..9194c1bbb9c 100644 --- a/pkg/kubelet/kubelet.go +++ b/pkg/kubelet/kubelet.go @@ -93,6 +93,7 @@ import ( kubetypes "k8s.io/kubernetes/pkg/kubelet/types" "k8s.io/kubernetes/pkg/kubelet/util/format" "k8s.io/kubernetes/pkg/kubelet/util/manager" + "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher" "k8s.io/kubernetes/pkg/kubelet/util/queue" "k8s.io/kubernetes/pkg/kubelet/util/sliceutils" "k8s.io/kubernetes/pkg/kubelet/volumemanager" @@ -775,6 +776,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration, if err != nil { return nil, err } + klet.pluginWatcher = pluginwatcher.NewWatcher(klet.getPluginsDir()) // If the experimentalMounterPathFlag is set, we do not want to // check node capabilities since the mount path is not the default @@ -1150,6 +1152,11 @@ type Kubelet struct { // This flag, if set, instructs the kubelet to keep volumes from terminated pods mounted to the node. // This can be useful for debugging volume related issues. keepTerminatedPodVolumes bool // DEPRECATED + + // pluginwatcher is a utility for Kubelet to register different types of node-level plugins + // such as device plugins or CSI plugins. It discovers plugins by monitoring inotify events under the + // directory returned by kubelet.getPluginsDir() + pluginWatcher pluginwatcher.Watcher } func allGlobalUnicastIPs() ([]net.IP, error) { @@ -1264,6 +1271,11 @@ func (kl *Kubelet) initializeModules() error { } } + // Start the plugin watcher + if err := kl.pluginWatcher.Start(); err != nil { + return fmt.Errorf("failed to start Plugin Watcher. err: %v", err) + } + // Start the image manager. kl.imageManager.Start() diff --git a/pkg/kubelet/util/BUILD b/pkg/kubelet/util/BUILD index df02ebdcd25..ff1755ebd11 100644 --- a/pkg/kubelet/util/BUILD +++ b/pkg/kubelet/util/BUILD @@ -93,9 +93,11 @@ filegroup( "//pkg/kubelet/util/format:all-srcs", "//pkg/kubelet/util/ioutils:all-srcs", "//pkg/kubelet/util/manager:all-srcs", + "//pkg/kubelet/util/pluginwatcher:all-srcs", "//pkg/kubelet/util/queue:all-srcs", "//pkg/kubelet/util/sliceutils:all-srcs", "//pkg/kubelet/util/store:all-srcs", ], tags = ["automanaged"], + visibility = ["//visibility:public"], ) diff --git a/pkg/kubelet/util/pluginwatcher/BUILD b/pkg/kubelet/util/pluginwatcher/BUILD new file mode 100644 index 00000000000..b4173ab5e1c --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/BUILD @@ -0,0 +1,58 @@ +package(default_visibility = ["//visibility:public"]) + +load( + "@io_bazel_rules_go//go:def.bzl", + "go_library", + "go_test", +) + +go_library( + name = "go_default_library", + srcs = [ + "example_plugin.go", + "plugin_watcher.go", + ], + importpath = "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher", + deps = [ + "//pkg/kubelet/apis/pluginregistration/v1alpha1:go_default_library", + "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1:go_default_library", + "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2:go_default_library", + "//pkg/util/filesystem:go_default_library", + "//vendor/github.com/fsnotify/fsnotify:go_default_library", + "//vendor/github.com/golang/glog:go_default_library", + "//vendor/golang.org/x/net/context:go_default_library", + "//vendor/google.golang.org/grpc:go_default_library", + ], +) + +filegroup( + name = "package-srcs", + srcs = glob(["**"]), + tags = ["automanaged"], + visibility = ["//visibility:private"], +) + +filegroup( + name = "all-srcs", + srcs = [ + ":package-srcs", + "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1:all-srcs", + "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2:all-srcs", + ], + tags = ["automanaged"], + visibility = ["//visibility:public"], +) + +go_test( + name = "go_default_test", + srcs = ["plugin_watcher_test.go"], + embed = [":go_default_library"], + deps = [ + "//pkg/kubelet/apis/pluginregistration/v1alpha1:go_default_library", + "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1:go_default_library", + "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2:go_default_library", + "//vendor/github.com/stretchr/testify/require:go_default_library", + "//vendor/golang.org/x/net/context:go_default_library", + "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library", + ], +) diff --git a/pkg/kubelet/util/pluginwatcher/README b/pkg/kubelet/util/pluginwatcher/README new file mode 100644 index 00000000000..9654b2cf62a --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/README @@ -0,0 +1,29 @@ +This folder contains a utility, pluginwatcher, for Kubelet to register +different types of node-level plugins such as device plugins or CSI plugins. +It discovers plugins by monitoring inotify events under the directory returned by +kubelet.getPluginsDir(). Lets refer this directory as PluginsSockDir. +For any discovered plugin, pluginwatcher issues Registration.GetInfo grpc call +to get plugin type, name and supported service API versions. For any registered plugin type, +pluginwatcher calls the registered callback function with the received plugin +name, supported service API versions, and the full socket path. The Kubelet +component that receives this callback can acknowledge or reject the plugin +according to its own logic, and use the socket path to establish its service +communication with any API version supported by the plugin. + +Here are the general rules that Kubelet plugin developers should follow: +- Run as 'root' user. Currently creating socket under PluginsSockDir, a root owned directory, requires + plugin process to be running as 'root'. +- Implements the Registration service specified in + pkg/kubelet/apis/pluginregistration/v*/api.proto. +- The plugin name sent during Registration.GetInfo grpc should be unique + for the given plugin type (CSIPlugin or DevicePlugin). +- The socket path needs to be unique and doesn't conflict with the path chosen + by any other potential plugins. Currently we only support flat fs namespace + under PluginsSockDir but will soon support recursive inotify watch for + hierarchical socket paths. +- A plugin should clean up its own socket upon exiting or when a new instance + comes up. A plugin should NOT remove any sockets belonging to other plugins. +- A plugin should make sure it has service ready for any supported service API + version listed in the PluginInfo. +- For an example plugin implementation, take a look at example_plugin.go + included in this directory. diff --git a/pkg/kubelet/util/pluginwatcher/example_plugin.go b/pkg/kubelet/util/pluginwatcher/example_plugin.go new file mode 100644 index 00000000000..fbca43acad5 --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/example_plugin.go @@ -0,0 +1,150 @@ +/* +Copyright 2018 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 pluginwatcher + +import ( + "fmt" + "net" + "sync" + "time" + + "github.com/golang/glog" + "golang.org/x/net/context" + "google.golang.org/grpc" + + registerapi "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1alpha1" + v1beta1 "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1" + v1beta2 "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2" +) + +const ( + PluginName = "example-plugin" + PluginType = "example-plugin-type" +) + +// examplePlugin is a sample plugin to work with plugin watcher +type examplePlugin struct { + grpcServer *grpc.Server + wg sync.WaitGroup + registrationStatus chan registerapi.RegistrationStatus // for testing + endpoint string // for testing +} + +type pluginServiceV1Beta1 struct { + server *examplePlugin +} + +func (s *pluginServiceV1Beta1) GetExampleInfo(ctx context.Context, rqt *v1beta1.ExampleRequest) (*v1beta1.ExampleResponse, error) { + glog.Infof("GetExampleInfo v1beta1field: %s", rqt.V1Beta1Field) + return &v1beta1.ExampleResponse{}, nil +} + +func (s *pluginServiceV1Beta1) RegisterService() { + v1beta1.RegisterExampleServer(s.server.grpcServer, s) +} + +type pluginServiceV1Beta2 struct { + server *examplePlugin +} + +func (s *pluginServiceV1Beta2) GetExampleInfo(ctx context.Context, rqt *v1beta2.ExampleRequest) (*v1beta2.ExampleResponse, error) { + glog.Infof("GetExampleInfo v1beta2_field: %s", rqt.V1Beta2Field) + return &v1beta2.ExampleResponse{}, nil +} + +func (s *pluginServiceV1Beta2) RegisterService() { + v1beta2.RegisterExampleServer(s.server.grpcServer, s) +} + +// NewExamplePlugin returns an initialized examplePlugin instance +func NewExamplePlugin() *examplePlugin { + return &examplePlugin{} +} + +// NewTestExamplePlugin returns an initialized examplePlugin instance for testing +func NewTestExamplePlugin(endpoint string) *examplePlugin { + return &examplePlugin{ + registrationStatus: make(chan registerapi.RegistrationStatus), + endpoint: endpoint, + } +} + +// GetInfo is the RPC invoked by plugin watcher +func (e *examplePlugin) GetInfo(ctx context.Context, req *registerapi.InfoRequest) (*registerapi.PluginInfo, error) { + return ®isterapi.PluginInfo{ + Type: PluginType, + Name: PluginName, + Endpoint: e.endpoint, + SupportedVersions: []string{"v1beta1", "v1beta2"}, + }, nil +} + +func (e *examplePlugin) NotifyRegistrationStatus(ctx context.Context, status *registerapi.RegistrationStatus) (*registerapi.RegistrationStatusResponse, error) { + if e.registrationStatus != nil { + e.registrationStatus <- *status + } + if !status.PluginRegistered { + glog.Errorf("Registration failed: %s\n", status.Error) + } + return ®isterapi.RegistrationStatusResponse{}, nil +} + +// Serve starts example plugin grpc server +func (e *examplePlugin) Serve(socketPath string) error { + glog.Infof("starting example server at: %s\n", socketPath) + lis, err := net.Listen("unix", socketPath) + if err != nil { + return err + } + glog.Infof("example server started at: %s\n", socketPath) + e.grpcServer = grpc.NewServer() + // Registers kubelet plugin watcher api. + registerapi.RegisterRegistrationServer(e.grpcServer, e) + // Registers services for both v1beta1 and v1beta2 versions. + v1beta1 := &pluginServiceV1Beta1{server: e} + v1beta1.RegisterService() + v1beta2 := &pluginServiceV1Beta2{server: e} + v1beta2.RegisterService() + + // Starts service + e.wg.Add(1) + go func() { + defer e.wg.Done() + // Blocking call to accept incoming connections. + if err := e.grpcServer.Serve(lis); err != nil { + glog.Errorf("example server stopped serving: %v", err) + } + }() + return nil +} + +func (e *examplePlugin) Stop() error { + glog.Infof("Stopping example server\n") + e.grpcServer.Stop() + c := make(chan struct{}) + go func() { + defer close(c) + e.wg.Wait() + }() + select { + case <-c: + return nil + case <-time.After(time.Second): + glog.Errorf("Timed out on waiting for stop completion") + return fmt.Errorf("Timed out on waiting for stop completion") + } +} diff --git a/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/BUILD b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/BUILD new file mode 100644 index 00000000000..affbd0aee4a --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/BUILD @@ -0,0 +1,34 @@ +load("@io_bazel_rules_go//go:def.bzl", "go_library") + +filegroup( + name = "go_default_library_protos", + srcs = ["api.proto"], + visibility = ["//visibility:public"], +) + +go_library( + name = "go_default_library", + srcs = ["api.pb.go"], + importpath = "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1", + visibility = ["//visibility:public"], + deps = [ + "//vendor/github.com/gogo/protobuf/gogoproto:go_default_library", + "//vendor/github.com/gogo/protobuf/proto:go_default_library", + "//vendor/golang.org/x/net/context:go_default_library", + "//vendor/google.golang.org/grpc:go_default_library", + ], +) + +filegroup( + name = "package-srcs", + srcs = glob(["**"]), + tags = ["automanaged"], + visibility = ["//visibility:private"], +) + +filegroup( + name = "all-srcs", + srcs = [":package-srcs"], + tags = ["automanaged"], + visibility = ["//visibility:public"], +) diff --git a/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/api.pb.go b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/api.pb.go new file mode 100644 index 00000000000..671e3df493b --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/api.pb.go @@ -0,0 +1,632 @@ +/* +Copyright 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. +*/ + +// Code generated by protoc-gen-gogo. +// source: api.proto +// DO NOT EDIT! + +/* + Package v1beta1 is a generated protocol buffer package. + + It is generated from these files: + api.proto + + It has these top-level messages: + ExampleRequest + ExampleResponse +*/ +package v1beta1 + +import proto "github.com/gogo/protobuf/proto" +import fmt "fmt" +import math "math" +import _ "github.com/gogo/protobuf/gogoproto" + +import ( + context "golang.org/x/net/context" + grpc "google.golang.org/grpc" +) + +import strings "strings" +import reflect "reflect" + +import io "io" + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package + +type ExampleRequest struct { + Request string `protobuf:"bytes,1,opt,name=request,proto3" json:"request,omitempty"` + V1Beta1Field string `protobuf:"bytes,2,opt,name=v1beta1_field,json=v1beta1Field,proto3" json:"v1beta1_field,omitempty"` +} + +func (m *ExampleRequest) Reset() { *m = ExampleRequest{} } +func (*ExampleRequest) ProtoMessage() {} +func (*ExampleRequest) Descriptor() ([]byte, []int) { return fileDescriptorApi, []int{0} } + +func (m *ExampleRequest) GetRequest() string { + if m != nil { + return m.Request + } + return "" +} + +func (m *ExampleRequest) GetV1Beta1Field() string { + if m != nil { + return m.V1Beta1Field + } + return "" +} + +type ExampleResponse struct { + Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"` +} + +func (m *ExampleResponse) Reset() { *m = ExampleResponse{} } +func (*ExampleResponse) ProtoMessage() {} +func (*ExampleResponse) Descriptor() ([]byte, []int) { return fileDescriptorApi, []int{1} } + +func (m *ExampleResponse) GetError() string { + if m != nil { + return m.Error + } + return "" +} + +func init() { + proto.RegisterType((*ExampleRequest)(nil), "v1beta1.ExampleRequest") + proto.RegisterType((*ExampleResponse)(nil), "v1beta1.ExampleResponse") +} + +// Reference imports to suppress errors if they are not otherwise used. +var _ context.Context +var _ grpc.ClientConn + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +const _ = grpc.SupportPackageIsVersion4 + +// Client API for Example service + +type ExampleClient interface { + GetExampleInfo(ctx context.Context, in *ExampleRequest, opts ...grpc.CallOption) (*ExampleResponse, error) +} + +type exampleClient struct { + cc *grpc.ClientConn +} + +func NewExampleClient(cc *grpc.ClientConn) ExampleClient { + return &exampleClient{cc} +} + +func (c *exampleClient) GetExampleInfo(ctx context.Context, in *ExampleRequest, opts ...grpc.CallOption) (*ExampleResponse, error) { + out := new(ExampleResponse) + err := grpc.Invoke(ctx, "/v1beta1.Example/GetExampleInfo", in, out, c.cc, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// Server API for Example service + +type ExampleServer interface { + GetExampleInfo(context.Context, *ExampleRequest) (*ExampleResponse, error) +} + +func RegisterExampleServer(s *grpc.Server, srv ExampleServer) { + s.RegisterService(&_Example_serviceDesc, srv) +} + +func _Example_GetExampleInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ExampleRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ExampleServer).GetExampleInfo(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/v1beta1.Example/GetExampleInfo", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ExampleServer).GetExampleInfo(ctx, req.(*ExampleRequest)) + } + return interceptor(ctx, in, info, handler) +} + +var _Example_serviceDesc = grpc.ServiceDesc{ + ServiceName: "v1beta1.Example", + HandlerType: (*ExampleServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "GetExampleInfo", + Handler: _Example_GetExampleInfo_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "api.proto", +} + +func (m *ExampleRequest) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *ExampleRequest) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Request) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.Request))) + i += copy(dAtA[i:], m.Request) + } + if len(m.V1Beta1Field) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.V1Beta1Field))) + i += copy(dAtA[i:], m.V1Beta1Field) + } + return i, nil +} + +func (m *ExampleResponse) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *ExampleResponse) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Error) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.Error))) + i += copy(dAtA[i:], m.Error) + } + return i, nil +} + +func encodeFixed64Api(dAtA []byte, offset int, v uint64) int { + dAtA[offset] = uint8(v) + dAtA[offset+1] = uint8(v >> 8) + dAtA[offset+2] = uint8(v >> 16) + dAtA[offset+3] = uint8(v >> 24) + dAtA[offset+4] = uint8(v >> 32) + dAtA[offset+5] = uint8(v >> 40) + dAtA[offset+6] = uint8(v >> 48) + dAtA[offset+7] = uint8(v >> 56) + return offset + 8 +} +func encodeFixed32Api(dAtA []byte, offset int, v uint32) int { + dAtA[offset] = uint8(v) + dAtA[offset+1] = uint8(v >> 8) + dAtA[offset+2] = uint8(v >> 16) + dAtA[offset+3] = uint8(v >> 24) + return offset + 4 +} +func encodeVarintApi(dAtA []byte, offset int, v uint64) int { + for v >= 1<<7 { + dAtA[offset] = uint8(v&0x7f | 0x80) + v >>= 7 + offset++ + } + dAtA[offset] = uint8(v) + return offset + 1 +} +func (m *ExampleRequest) Size() (n int) { + var l int + _ = l + l = len(m.Request) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + l = len(m.V1Beta1Field) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + return n +} + +func (m *ExampleResponse) Size() (n int) { + var l int + _ = l + l = len(m.Error) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + return n +} + +func sovApi(x uint64) (n int) { + for { + n++ + x >>= 7 + if x == 0 { + break + } + } + return n +} +func sozApi(x uint64) (n int) { + return sovApi(uint64((x << 1) ^ uint64((int64(x) >> 63)))) +} +func (this *ExampleRequest) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&ExampleRequest{`, + `Request:` + fmt.Sprintf("%v", this.Request) + `,`, + `V1Beta1Field:` + fmt.Sprintf("%v", this.V1Beta1Field) + `,`, + `}`, + }, "") + return s +} +func (this *ExampleResponse) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&ExampleResponse{`, + `Error:` + fmt.Sprintf("%v", this.Error) + `,`, + `}`, + }, "") + return s +} +func valueToStringApi(v interface{}) string { + rv := reflect.ValueOf(v) + if rv.IsNil() { + return "nil" + } + pv := reflect.Indirect(rv).Interface() + return fmt.Sprintf("*%v", pv) +} +func (m *ExampleRequest) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: ExampleRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: ExampleRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Request", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Request = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field V1Beta1Field", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.V1Beta1Field = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipApi(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthApi + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *ExampleResponse) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: ExampleResponse: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: ExampleResponse: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Error", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Error = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipApi(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthApi + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func skipApi(dAtA []byte) (n int, err error) { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + wireType := int(wire & 0x7) + switch wireType { + case 0: + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + iNdEx++ + if dAtA[iNdEx-1] < 0x80 { + break + } + } + return iNdEx, nil + case 1: + iNdEx += 8 + return iNdEx, nil + case 2: + var length int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + length |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + iNdEx += length + if length < 0 { + return 0, ErrInvalidLengthApi + } + return iNdEx, nil + case 3: + for { + var innerWire uint64 + var start int = iNdEx + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + innerWire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + innerWireType := int(innerWire & 0x7) + if innerWireType == 4 { + break + } + next, err := skipApi(dAtA[start:]) + if err != nil { + return 0, err + } + iNdEx = start + next + } + return iNdEx, nil + case 4: + return iNdEx, nil + case 5: + iNdEx += 4 + return iNdEx, nil + default: + return 0, fmt.Errorf("proto: illegal wireType %d", wireType) + } + } + panic("unreachable") +} + +var ( + ErrInvalidLengthApi = fmt.Errorf("proto: negative length found during unmarshaling") + ErrIntOverflowApi = fmt.Errorf("proto: integer overflow") +) + +func init() { proto.RegisterFile("api.proto", fileDescriptorApi) } + +var fileDescriptorApi = []byte{ + // 227 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4c, 0x2c, 0xc8, 0xd4, + 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x2f, 0x33, 0x4c, 0x4a, 0x2d, 0x49, 0x34, 0x94, 0xd2, + 0x4d, 0xcf, 0x2c, 0xc9, 0x28, 0x4d, 0xd2, 0x4b, 0xce, 0xcf, 0xd5, 0x4f, 0xcf, 0x4f, 0xcf, 0xd7, + 0x07, 0xcb, 0x27, 0x95, 0xa6, 0x81, 0x79, 0x60, 0x0e, 0x98, 0x05, 0xd1, 0xa7, 0xe4, 0xcf, 0xc5, + 0xe7, 0x5a, 0x91, 0x98, 0x5b, 0x90, 0x93, 0x1a, 0x94, 0x5a, 0x58, 0x9a, 0x5a, 0x5c, 0x22, 0x24, + 0xc1, 0xc5, 0x5e, 0x04, 0x61, 0x4a, 0x30, 0x2a, 0x30, 0x6a, 0x70, 0x06, 0xc1, 0xb8, 0x42, 0xca, + 0x5c, 0xbc, 0x50, 0x5b, 0xe2, 0xd3, 0x32, 0x53, 0x73, 0x52, 0x24, 0x98, 0xc0, 0xf2, 0x3c, 0x50, + 0x41, 0x37, 0x90, 0x98, 0x92, 0x3a, 0x17, 0x3f, 0xdc, 0xc0, 0xe2, 0x82, 0xfc, 0xbc, 0xe2, 0x54, + 0x21, 0x11, 0x2e, 0xd6, 0xd4, 0xa2, 0xa2, 0xfc, 0x22, 0xa8, 0x79, 0x10, 0x8e, 0x51, 0x00, 0x17, + 0x3b, 0x54, 0xa1, 0x90, 0x2b, 0x17, 0x9f, 0x7b, 0x6a, 0x09, 0x94, 0xe7, 0x99, 0x97, 0x96, 0x2f, + 0x24, 0xae, 0x07, 0x35, 0x54, 0x0f, 0xd5, 0x75, 0x52, 0x12, 0x98, 0x12, 0x10, 0x5b, 0x94, 0x18, + 0x9c, 0x64, 0x4e, 0x3c, 0x94, 0x63, 0xbc, 0xf1, 0x50, 0x8e, 0xa1, 0xe1, 0x91, 0x1c, 0xe3, 0x89, + 0x47, 0x72, 0x8c, 0x17, 0x1e, 0xc9, 0x31, 0x3e, 0x78, 0x24, 0xc7, 0x38, 0xe1, 0xb1, 0x1c, 0x43, + 0x12, 0x1b, 0xd8, 0xc3, 0xc6, 0x80, 0x00, 0x00, 0x00, 0xff, 0xff, 0x99, 0x62, 0xd1, 0x9c, 0x35, + 0x01, 0x00, 0x00, +} diff --git a/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/api.proto b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/api.proto new file mode 100644 index 00000000000..14aa7df2c4d --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/api.proto @@ -0,0 +1,28 @@ +syntax = 'proto3'; + +package v1beta1; + +import "github.com/gogo/protobuf/gogoproto/gogo.proto"; + +option (gogoproto.goproto_stringer_all) = false; +option (gogoproto.stringer_all) = true; +option (gogoproto.goproto_getters_all) = true; +option (gogoproto.marshaler_all) = true; +option (gogoproto.sizer_all) = true; +option (gogoproto.unmarshaler_all) = true; +option (gogoproto.goproto_unrecognized_all) = false; + +message ExampleRequest { + string request = 1; + string v1beta1_field = 2; +} + +message ExampleResponse { + string error = 1; +} + +// Example is a simple example service for general reference on the recommended +// kubelet plugin model and plugin watcher testing. +service Example { + rpc GetExampleInfo(ExampleRequest) returns (ExampleResponse) {} +} diff --git a/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/BUILD b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/BUILD new file mode 100644 index 00000000000..f2b53898d38 --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/BUILD @@ -0,0 +1,34 @@ +load("@io_bazel_rules_go//go:def.bzl", "go_library") + +filegroup( + name = "go_default_library_protos", + srcs = ["api.proto"], + visibility = ["//visibility:public"], +) + +go_library( + name = "go_default_library", + srcs = ["api.pb.go"], + importpath = "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2", + visibility = ["//visibility:public"], + deps = [ + "//vendor/github.com/gogo/protobuf/gogoproto:go_default_library", + "//vendor/github.com/gogo/protobuf/proto:go_default_library", + "//vendor/golang.org/x/net/context:go_default_library", + "//vendor/google.golang.org/grpc:go_default_library", + ], +) + +filegroup( + name = "package-srcs", + srcs = glob(["**"]), + tags = ["automanaged"], + visibility = ["//visibility:private"], +) + +filegroup( + name = "all-srcs", + srcs = [":package-srcs"], + tags = ["automanaged"], + visibility = ["//visibility:public"], +) diff --git a/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/api.pb.go b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/api.pb.go new file mode 100644 index 00000000000..0c63b31429f --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/api.pb.go @@ -0,0 +1,633 @@ +/* +Copyright 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. +*/ + +// Code generated by protoc-gen-gogo. +// source: api.proto +// DO NOT EDIT! + +/* + Package v1beta2 is a generated protocol buffer package. + + It is generated from these files: + api.proto + + It has these top-level messages: + ExampleRequest + ExampleResponse +*/ +package v1beta2 + +import proto "github.com/gogo/protobuf/proto" +import fmt "fmt" +import math "math" +import _ "github.com/gogo/protobuf/gogoproto" + +import ( + context "golang.org/x/net/context" + grpc "google.golang.org/grpc" +) + +import strings "strings" +import reflect "reflect" + +import io "io" + +// Reference imports to suppress errors if they are not otherwise used. +var _ = proto.Marshal +var _ = fmt.Errorf +var _ = math.Inf + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the proto package it is being compiled against. +// A compilation error at this line likely means your copy of the +// proto package needs to be updated. +const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package + +// Renames a field from v1beta1 ExampleRequest. +type ExampleRequest struct { + Request string `protobuf:"bytes,1,opt,name=request,proto3" json:"request,omitempty"` + V1Beta2Field string `protobuf:"bytes,2,opt,name=v1beta2_field,json=v1beta2Field,proto3" json:"v1beta2_field,omitempty"` +} + +func (m *ExampleRequest) Reset() { *m = ExampleRequest{} } +func (*ExampleRequest) ProtoMessage() {} +func (*ExampleRequest) Descriptor() ([]byte, []int) { return fileDescriptorApi, []int{0} } + +func (m *ExampleRequest) GetRequest() string { + if m != nil { + return m.Request + } + return "" +} + +func (m *ExampleRequest) GetV1Beta2Field() string { + if m != nil { + return m.V1Beta2Field + } + return "" +} + +type ExampleResponse struct { + Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"` +} + +func (m *ExampleResponse) Reset() { *m = ExampleResponse{} } +func (*ExampleResponse) ProtoMessage() {} +func (*ExampleResponse) Descriptor() ([]byte, []int) { return fileDescriptorApi, []int{1} } + +func (m *ExampleResponse) GetError() string { + if m != nil { + return m.Error + } + return "" +} + +func init() { + proto.RegisterType((*ExampleRequest)(nil), "v1beta2.ExampleRequest") + proto.RegisterType((*ExampleResponse)(nil), "v1beta2.ExampleResponse") +} + +// Reference imports to suppress errors if they are not otherwise used. +var _ context.Context +var _ grpc.ClientConn + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +const _ = grpc.SupportPackageIsVersion4 + +// Client API for Example service + +type ExampleClient interface { + GetExampleInfo(ctx context.Context, in *ExampleRequest, opts ...grpc.CallOption) (*ExampleResponse, error) +} + +type exampleClient struct { + cc *grpc.ClientConn +} + +func NewExampleClient(cc *grpc.ClientConn) ExampleClient { + return &exampleClient{cc} +} + +func (c *exampleClient) GetExampleInfo(ctx context.Context, in *ExampleRequest, opts ...grpc.CallOption) (*ExampleResponse, error) { + out := new(ExampleResponse) + err := grpc.Invoke(ctx, "/v1beta2.Example/GetExampleInfo", in, out, c.cc, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// Server API for Example service + +type ExampleServer interface { + GetExampleInfo(context.Context, *ExampleRequest) (*ExampleResponse, error) +} + +func RegisterExampleServer(s *grpc.Server, srv ExampleServer) { + s.RegisterService(&_Example_serviceDesc, srv) +} + +func _Example_GetExampleInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ExampleRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ExampleServer).GetExampleInfo(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/v1beta2.Example/GetExampleInfo", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ExampleServer).GetExampleInfo(ctx, req.(*ExampleRequest)) + } + return interceptor(ctx, in, info, handler) +} + +var _Example_serviceDesc = grpc.ServiceDesc{ + ServiceName: "v1beta2.Example", + HandlerType: (*ExampleServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "GetExampleInfo", + Handler: _Example_GetExampleInfo_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "api.proto", +} + +func (m *ExampleRequest) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *ExampleRequest) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Request) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.Request))) + i += copy(dAtA[i:], m.Request) + } + if len(m.V1Beta2Field) > 0 { + dAtA[i] = 0x12 + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.V1Beta2Field))) + i += copy(dAtA[i:], m.V1Beta2Field) + } + return i, nil +} + +func (m *ExampleResponse) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *ExampleResponse) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Error) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintApi(dAtA, i, uint64(len(m.Error))) + i += copy(dAtA[i:], m.Error) + } + return i, nil +} + +func encodeFixed64Api(dAtA []byte, offset int, v uint64) int { + dAtA[offset] = uint8(v) + dAtA[offset+1] = uint8(v >> 8) + dAtA[offset+2] = uint8(v >> 16) + dAtA[offset+3] = uint8(v >> 24) + dAtA[offset+4] = uint8(v >> 32) + dAtA[offset+5] = uint8(v >> 40) + dAtA[offset+6] = uint8(v >> 48) + dAtA[offset+7] = uint8(v >> 56) + return offset + 8 +} +func encodeFixed32Api(dAtA []byte, offset int, v uint32) int { + dAtA[offset] = uint8(v) + dAtA[offset+1] = uint8(v >> 8) + dAtA[offset+2] = uint8(v >> 16) + dAtA[offset+3] = uint8(v >> 24) + return offset + 4 +} +func encodeVarintApi(dAtA []byte, offset int, v uint64) int { + for v >= 1<<7 { + dAtA[offset] = uint8(v&0x7f | 0x80) + v >>= 7 + offset++ + } + dAtA[offset] = uint8(v) + return offset + 1 +} +func (m *ExampleRequest) Size() (n int) { + var l int + _ = l + l = len(m.Request) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + l = len(m.V1Beta2Field) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + return n +} + +func (m *ExampleResponse) Size() (n int) { + var l int + _ = l + l = len(m.Error) + if l > 0 { + n += 1 + l + sovApi(uint64(l)) + } + return n +} + +func sovApi(x uint64) (n int) { + for { + n++ + x >>= 7 + if x == 0 { + break + } + } + return n +} +func sozApi(x uint64) (n int) { + return sovApi(uint64((x << 1) ^ uint64((int64(x) >> 63)))) +} +func (this *ExampleRequest) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&ExampleRequest{`, + `Request:` + fmt.Sprintf("%v", this.Request) + `,`, + `V1Beta2Field:` + fmt.Sprintf("%v", this.V1Beta2Field) + `,`, + `}`, + }, "") + return s +} +func (this *ExampleResponse) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&ExampleResponse{`, + `Error:` + fmt.Sprintf("%v", this.Error) + `,`, + `}`, + }, "") + return s +} +func valueToStringApi(v interface{}) string { + rv := reflect.ValueOf(v) + if rv.IsNil() { + return "nil" + } + pv := reflect.Indirect(rv).Interface() + return fmt.Sprintf("*%v", pv) +} +func (m *ExampleRequest) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: ExampleRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: ExampleRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Request", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Request = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field V1Beta2Field", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.V1Beta2Field = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipApi(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthApi + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *ExampleResponse) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: ExampleResponse: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: ExampleResponse: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Error", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowApi + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthApi + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Error = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipApi(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthApi + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func skipApi(dAtA []byte) (n int, err error) { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + wireType := int(wire & 0x7) + switch wireType { + case 0: + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + iNdEx++ + if dAtA[iNdEx-1] < 0x80 { + break + } + } + return iNdEx, nil + case 1: + iNdEx += 8 + return iNdEx, nil + case 2: + var length int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + length |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + iNdEx += length + if length < 0 { + return 0, ErrInvalidLengthApi + } + return iNdEx, nil + case 3: + for { + var innerWire uint64 + var start int = iNdEx + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return 0, ErrIntOverflowApi + } + if iNdEx >= l { + return 0, io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + innerWire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + innerWireType := int(innerWire & 0x7) + if innerWireType == 4 { + break + } + next, err := skipApi(dAtA[start:]) + if err != nil { + return 0, err + } + iNdEx = start + next + } + return iNdEx, nil + case 4: + return iNdEx, nil + case 5: + iNdEx += 4 + return iNdEx, nil + default: + return 0, fmt.Errorf("proto: illegal wireType %d", wireType) + } + } + panic("unreachable") +} + +var ( + ErrInvalidLengthApi = fmt.Errorf("proto: negative length found during unmarshaling") + ErrIntOverflowApi = fmt.Errorf("proto: integer overflow") +) + +func init() { proto.RegisterFile("api.proto", fileDescriptorApi) } + +var fileDescriptorApi = []byte{ + // 227 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4c, 0x2c, 0xc8, 0xd4, + 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x2f, 0x33, 0x4c, 0x4a, 0x2d, 0x49, 0x34, 0x92, 0xd2, + 0x4d, 0xcf, 0x2c, 0xc9, 0x28, 0x4d, 0xd2, 0x4b, 0xce, 0xcf, 0xd5, 0x4f, 0xcf, 0x4f, 0xcf, 0xd7, + 0x07, 0xcb, 0x27, 0x95, 0xa6, 0x81, 0x79, 0x60, 0x0e, 0x98, 0x05, 0xd1, 0xa7, 0xe4, 0xcf, 0xc5, + 0xe7, 0x5a, 0x91, 0x98, 0x5b, 0x90, 0x93, 0x1a, 0x94, 0x5a, 0x58, 0x9a, 0x5a, 0x5c, 0x22, 0x24, + 0xc1, 0xc5, 0x5e, 0x04, 0x61, 0x4a, 0x30, 0x2a, 0x30, 0x6a, 0x70, 0x06, 0xc1, 0xb8, 0x42, 0xca, + 0x5c, 0xbc, 0x50, 0x5b, 0xe2, 0xd3, 0x32, 0x53, 0x73, 0x52, 0x24, 0x98, 0xc0, 0xf2, 0x3c, 0x50, + 0x41, 0x37, 0x90, 0x98, 0x92, 0x3a, 0x17, 0x3f, 0xdc, 0xc0, 0xe2, 0x82, 0xfc, 0xbc, 0xe2, 0x54, + 0x21, 0x11, 0x2e, 0xd6, 0xd4, 0xa2, 0xa2, 0xfc, 0x22, 0xa8, 0x79, 0x10, 0x8e, 0x51, 0x00, 0x17, + 0x3b, 0x54, 0xa1, 0x90, 0x2b, 0x17, 0x9f, 0x7b, 0x6a, 0x09, 0x94, 0xe7, 0x99, 0x97, 0x96, 0x2f, + 0x24, 0xae, 0x07, 0x35, 0x54, 0x0f, 0xd5, 0x75, 0x52, 0x12, 0x98, 0x12, 0x10, 0x5b, 0x94, 0x18, + 0x9c, 0x64, 0x4e, 0x3c, 0x94, 0x63, 0xbc, 0xf1, 0x50, 0x8e, 0xa1, 0xe1, 0x91, 0x1c, 0xe3, 0x89, + 0x47, 0x72, 0x8c, 0x17, 0x1e, 0xc9, 0x31, 0x3e, 0x78, 0x24, 0xc7, 0x38, 0xe1, 0xb1, 0x1c, 0x43, + 0x12, 0x1b, 0xd8, 0xc3, 0xc6, 0x80, 0x00, 0x00, 0x00, 0xff, 0xff, 0xa8, 0x79, 0x17, 0x13, 0x35, + 0x01, 0x00, 0x00, +} diff --git a/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/api.proto b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/api.proto new file mode 100644 index 00000000000..e34697f3a66 --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/api.proto @@ -0,0 +1,29 @@ +syntax = 'proto3'; + +package v1beta2; + +import "github.com/gogo/protobuf/gogoproto/gogo.proto"; + +option (gogoproto.goproto_stringer_all) = false; +option (gogoproto.stringer_all) = true; +option (gogoproto.goproto_getters_all) = true; +option (gogoproto.marshaler_all) = true; +option (gogoproto.sizer_all) = true; +option (gogoproto.unmarshaler_all) = true; +option (gogoproto.goproto_unrecognized_all) = false; + +// Renames a field from v1beta1 ExampleRequest. +message ExampleRequest { + string request = 1; + string v1beta2_field = 2; +} + +message ExampleResponse { + string error = 1; +} + +// Example is a simple example service for general reference on the recommended +// kubelet plugin model and plugin watcher testing. +service Example { + rpc GetExampleInfo(ExampleRequest) returns (ExampleResponse) {} +} diff --git a/pkg/kubelet/util/pluginwatcher/plugin_watcher.go b/pkg/kubelet/util/pluginwatcher/plugin_watcher.go new file mode 100644 index 00000000000..9a5241cb2e5 --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/plugin_watcher.go @@ -0,0 +1,260 @@ +/* +Copyright 2018 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 pluginwatcher + +import ( + "fmt" + "net" + "os" + "path" + "path/filepath" + "sync" + "time" + + "github.com/fsnotify/fsnotify" + "github.com/golang/glog" + "golang.org/x/net/context" + "google.golang.org/grpc" + registerapi "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1alpha1" + utilfs "k8s.io/kubernetes/pkg/util/filesystem" +) + +// RegisterCallbackFn is the type of the callback function that handlers will provide +type RegisterCallbackFn func(pluginName string, endpoint string, versions []string, socketPath string) (error, chan bool) + +// Watcher is the plugin watcher +type Watcher struct { + path string + handlers map[string]RegisterCallbackFn + stopCh chan interface{} + fs utilfs.Filesystem + watcher *fsnotify.Watcher + wg sync.WaitGroup + mutex sync.Mutex +} + +// NewWatcher provides a new watcher +func NewWatcher(sockDir string) Watcher { + return Watcher{ + path: sockDir, + handlers: make(map[string]RegisterCallbackFn), + fs: &utilfs.DefaultFs{}, + } +} + +// AddHandler registers a callback to be invoked for a particular type of plugin +func (w *Watcher) AddHandler(handlerType string, handlerCbkFn RegisterCallbackFn) { + w.mutex.Lock() + defer w.mutex.Unlock() + w.handlers[handlerType] = handlerCbkFn +} + +// Creates the plugin directory, if it doesn't already exist. +func (w *Watcher) createPluginDir() error { + glog.V(4).Infof("Ensuring Plugin directory at %s ", w.path) + if err := w.fs.MkdirAll(w.path, 0755); err != nil { + return fmt.Errorf("error (re-)creating driver directory: %s", err) + } + return nil +} + +// Walks through the plugin directory to discover any existing plugin sockets. +func (w *Watcher) traversePluginDir() error { + files, err := w.fs.ReadDir(w.path) + if err != nil { + return fmt.Errorf("error reading the plugin directory: %v", err) + } + for _, f := range files { + // Currently only supports flat fs namespace under the plugin directory. + // TODO: adds support for hierarchical fs namespace. + if !f.IsDir() && filepath.Base(f.Name())[0] != '.' { + go func(sockName string) { + w.watcher.Events <- fsnotify.Event{ + Name: sockName, + Op: fsnotify.Op(uint32(1)), + } + }(path.Join(w.path, f.Name())) + } + } + return nil +} + +func (w *Watcher) init() error { + if err := w.createPluginDir(); err != nil { + return err + } + return nil +} + +func (w *Watcher) registerPlugin(socketPath string) error { + //TODO: Implement rate limiting to mitigate any DOS kind of attacks. + glog.V(4).Infof("registerPlugin called for socketPath: %s", socketPath) + client, conn, err := dial(socketPath) + if err != nil { + return fmt.Errorf("dial failed at socket %s, err: %v", socketPath, err) + } + defer conn.Close() + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + infoResp, err := client.GetInfo(ctx, ®isterapi.InfoRequest{}) + if err != nil { + return fmt.Errorf("failed to get plugin info using RPC GetInfo at socket %s, err: %v", socketPath, err) + } + if err := w.invokeRegistrationCallbackAtHandler(ctx, client, infoResp, socketPath); err != nil { + return fmt.Errorf("failed to register plugin. Callback handler returned err: %v", err) + } + glog.V(4).Infof("Successfully registered plugin for plugin type: %s, name: %s, socket: %s", infoResp.Type, infoResp.Name, socketPath) + return nil +} + +func (w *Watcher) invokeRegistrationCallbackAtHandler(ctx context.Context, client registerapi.RegistrationClient, infoResp *registerapi.PluginInfo, socketPath string) error { + var handlerCbkFn RegisterCallbackFn + var ok bool + handlerCbkFn, ok = w.handlers[infoResp.Type] + if !ok { + if _, err := client.NotifyRegistrationStatus(ctx, ®isterapi.RegistrationStatus{ + PluginRegistered: false, + Error: fmt.Sprintf("No handler found registered for plugin type: %s, socket: %s", infoResp.Type, socketPath), + }); err != nil { + glog.Errorf("Failed to send registration status at socket %s, err: %v", socketPath, err) + } + return fmt.Errorf("no handler found registered for plugin type: %s, socket: %s", infoResp.Type, socketPath) + } + + var versions []string + for _, version := range infoResp.SupportedVersions { + versions = append(versions, version) + } + // calls handler callback to verify registration request + err, chanForAckOfNotification := handlerCbkFn(infoResp.Name, infoResp.Endpoint, versions, socketPath) + if err != nil { + if _, err := client.NotifyRegistrationStatus(ctx, ®isterapi.RegistrationStatus{ + PluginRegistered: false, + Error: fmt.Sprintf("Plugin registration failed with err: %v", err), + }); err != nil { + glog.Errorf("Failed to send registration status at socket %s, err: %v", socketPath, err) + } + chanForAckOfNotification <- false + return fmt.Errorf("plugin registration failed with err: %v", err) + } + + if _, err := client.NotifyRegistrationStatus(ctx, ®isterapi.RegistrationStatus{ + PluginRegistered: true, + }); err != nil { + return fmt.Errorf("failed to send registration status at socket %s, err: %v", socketPath, err) + } + chanForAckOfNotification <- true + return nil +} + +// Start watches for the creation of plugin sockets at the path +func (w *Watcher) Start() error { + glog.V(2).Infof("Plugin Watcher Start at %s", w.path) + w.stopCh = make(chan interface{}) + + // Creating the directory to be watched if it doesn't exist yet, + // and walks through the directory to discover the existing plugins. + if err := w.init(); err != nil { + return err + } + + watcher, err := fsnotify.NewWatcher() + if err != nil { + return fmt.Errorf("failed to start plugin watcher, err: %v", err) + } + + if err := watcher.Add(w.path); err != nil { + watcher.Close() + return fmt.Errorf("failed to start plugin watcher, err: %v", err) + } + + w.watcher = watcher + + if err := w.traversePluginDir(); err != nil { + watcher.Close() + return fmt.Errorf("failed to traverse plugin socket path, err: %v", err) + } + + w.wg.Add(1) + go func(watcher *fsnotify.Watcher) { + defer w.wg.Done() + for { + select { + case event := <-watcher.Events: + if event.Op&fsnotify.Create == fsnotify.Create { + go func(eventName string) { + err := w.registerPlugin(eventName) + if err != nil { + glog.Errorf("Plugin %s registration failed with error: %v", eventName, err) + } + }(event.Name) + } + continue + case err := <-watcher.Errors: + //TODO: Handle errors by taking corrective measures + if err != nil { + glog.Errorf("Watcher received error: %v", err) + } + continue + + case <-w.stopCh: + watcher.Close() + break + } + break + } + }(watcher) + return nil +} + +// Stop stops probing the creation of plugin sockets at the path +func (w *Watcher) Stop() error { + close(w.stopCh) + c := make(chan struct{}) + go func() { + defer close(c) + w.wg.Wait() + }() + select { + case <-c: + case <-time.After(10 * time.Second): + return fmt.Errorf("timeout on stopping watcher") + } + return nil +} + +// Cleanup cleans the path by removing sockets +func (w *Watcher) Cleanup() error { + return os.RemoveAll(w.path) +} + +// Dial establishes the gRPC communication with the picked up plugin socket. https://godoc.org/google.golang.org/grpc#Dial +func dial(unixSocketPath string) (registerapi.RegistrationClient, *grpc.ClientConn, error) { + c, err := grpc.Dial(unixSocketPath, grpc.WithInsecure(), grpc.WithBlock(), + grpc.WithTimeout(10*time.Second), + grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) { + return net.DialTimeout("unix", addr, timeout) + }), + ) + + if err != nil { + return nil, nil, fmt.Errorf("failed to dial socket %s, err: %v", unixSocketPath, err) + } + + return registerapi.NewRegistrationClient(c), c, nil +} diff --git a/pkg/kubelet/util/pluginwatcher/plugin_watcher_test.go b/pkg/kubelet/util/pluginwatcher/plugin_watcher_test.go new file mode 100644 index 00000000000..44bccf9a6f3 --- /dev/null +++ b/pkg/kubelet/util/pluginwatcher/plugin_watcher_test.go @@ -0,0 +1,220 @@ +/* +Copyright 2018 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 pluginwatcher + +import ( + "fmt" + "io/ioutil" + "strconv" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "golang.org/x/net/context" + + "k8s.io/apimachinery/pkg/util/sets" + registerapi "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1alpha1" + v1beta1 "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1" + v1beta2 "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2" +) + +func TestExamplePlugin(t *testing.T) { + socketDir, err := ioutil.TempDir("", "plugin_test") + require.NoError(t, err) + socketPath := socketDir + "/plugin.sock" + w := NewWatcher(socketDir) + + testCases := []struct { + description string + expectedEndpoint string + returnErr error + }{ + { + description: "Successfully register plugin through inotify", + expectedEndpoint: "", + returnErr: nil, + }, + { + description: "Successfully register plugin through inotify and got expected optional endpoint", + expectedEndpoint: "dummyEndpoint", + returnErr: nil, + }, + { + description: "Fails registration because endpoint is expected to be non-empty", + expectedEndpoint: "dummyEndpoint", + returnErr: fmt.Errorf("empty endpoint received"), + }, + { + description: "Successfully register plugin through inotify after plugin restarts", + expectedEndpoint: "", + returnErr: nil, + }, + { + description: "Fails registration with conflicting plugin name", + expectedEndpoint: "", + returnErr: fmt.Errorf("conflicting plugin name"), + }, + { + description: "Successfully register plugin during initial traverse after plugin watcher restarts", + expectedEndpoint: "", + returnErr: nil, + }, + { + description: "Fails registration with conflicting plugin name during initial traverse after plugin watcher restarts", + expectedEndpoint: "", + returnErr: fmt.Errorf("conflicting plugin name"), + }, + } + + callbackCount := struct { + mutex sync.Mutex + count int32 + }{} + w.AddHandler(PluginType, func(name string, endpoint string, versions []string, sockPath string) (error, chan bool) { + callbackCount.mutex.Lock() + localCount := callbackCount.count + callbackCount.count = callbackCount.count + 1 + callbackCount.mutex.Unlock() + + require.True(t, localCount <= int32((len(testCases)-1))) + require.Equal(t, PluginName, name, "Plugin name mismatched!!") + retError := testCases[localCount].returnErr + if retError == nil || retError.Error() != "empty endpoint received" { + require.Equal(t, testCases[localCount].expectedEndpoint, endpoint, "Unexpected endpoint") + } else { + require.NotEqual(t, testCases[localCount].expectedEndpoint, endpoint, "Unexpected endpoint") + } + + require.Equal(t, []string{"v1beta1", "v1beta2"}, versions, "Plugin version mismatched!!") + // Verifies the grpcServer is ready to serve services. + _, conn, err := dial(sockPath) + require.Nil(t, err) + defer conn.Close() + + // The plugin handler should be able to use any listed service API version. + v1beta1Client := v1beta1.NewExampleClient(conn) + v1beta2Client := v1beta2.NewExampleClient(conn) + + // Tests v1beta1 GetExampleInfo + _, err = v1beta1Client.GetExampleInfo(context.Background(), &v1beta1.ExampleRequest{}) + require.Nil(t, err) + + // Tests v1beta1 GetExampleInfo + _, err = v1beta2Client.GetExampleInfo(context.Background(), &v1beta2.ExampleRequest{}) + //atomic.AddInt32(&callbackCount, 1) + chanForAckOfNotification := make(chan bool) + + go func() { + select { + case <-chanForAckOfNotification: + close(chanForAckOfNotification) + case <-time.After(time.Second): + t.Fatalf("Timed out while waiting for notification ack") + } + }() + return retError, chanForAckOfNotification + }) + require.NoError(t, w.Start()) + + p := NewTestExamplePlugin("") + require.NoError(t, p.Serve(socketPath)) + require.True(t, waitForPluginRegistrationStatus(t, p.registrationStatus)) + + require.NoError(t, p.Stop()) + + p = NewTestExamplePlugin("dummyEndpoint") + require.NoError(t, p.Serve(socketPath)) + require.True(t, waitForPluginRegistrationStatus(t, p.registrationStatus)) + + require.NoError(t, p.Stop()) + + p = NewTestExamplePlugin("") + require.NoError(t, p.Serve(socketPath)) + require.False(t, waitForPluginRegistrationStatus(t, p.registrationStatus)) + + // Trying to start a plugin service at the same socket path should fail + // with "bind: address already in use" + require.NotNil(t, p.Serve(socketPath)) + + // grpcServer.Stop() will remove the socket and starting plugin service + // at the same path again should succeeds and trigger another callback. + require.NoError(t, p.Stop()) + p = NewTestExamplePlugin("") + go func() { + require.Nil(t, p.Serve(socketPath)) + }() + require.True(t, waitForPluginRegistrationStatus(t, p.registrationStatus)) + + // Starting another plugin with the same name got verification error. + p2 := NewTestExamplePlugin("") + socketPath2 := socketDir + "/plugin2.sock" + go func() { + require.NoError(t, p2.Serve(socketPath2)) + }() + require.False(t, waitForPluginRegistrationStatus(t, p2.registrationStatus)) + + // Restarts plugin watcher should traverse the socket directory and issues a + // callback for every existing socket. + require.NoError(t, w.Stop()) + errCh := make(chan error) + go func() { + errCh <- w.Start() + }() + + var wg sync.WaitGroup + wg.Add(2) + var pStatus string + var p2Status string + go func() { + pStatus = strconv.FormatBool(waitForPluginRegistrationStatus(t, p.registrationStatus)) + wg.Done() + }() + go func() { + p2Status = strconv.FormatBool(waitForPluginRegistrationStatus(t, p2.registrationStatus)) + wg.Done() + }() + wg.Wait() + expectedSet := sets.NewString() + expectedSet.Insert("true", "false") + actualSet := sets.NewString() + actualSet.Insert(pStatus, p2Status) + + require.Equal(t, expectedSet, actualSet) + + select { + case err = <-errCh: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatalf("Timed out while waiting for watcher start") + + } + + require.NoError(t, w.Stop()) + err = w.Cleanup() + require.NoError(t, err) +} + +func waitForPluginRegistrationStatus(t *testing.T, statusCh chan registerapi.RegistrationStatus) bool { + select { + case status := <-statusCh: + return status.PluginRegistered + case <-time.After(10 * time.Second): + t.Fatalf("Timed out while waiting for registration status") + } + return false +}