diff --git a/core/dnsserver/server_grpc.go b/core/dnsserver/server_grpc.go index 7de36a5fd..5a1dcc709 100644 --- a/core/dnsserver/server_grpc.go +++ b/core/dnsserver/server_grpc.go @@ -9,7 +9,6 @@ import ( "github.com/coredns/coredns/pb" "github.com/coredns/coredns/plugin/pkg/transport" - "github.com/coredns/coredns/plugin/pkg/watch" "github.com/grpc-ecosystem/grpc-opentracing/go/otgrpc" "github.com/miekg/dns" @@ -24,7 +23,6 @@ type ServergRPC struct { grpcServer *grpc.Server listenAddr net.Addr tlsConfig *tls.Config - watch watch.Watcher } // NewServergRPC returns a new CoreDNS GRPC server and compiles all plugin in to it. @@ -41,7 +39,7 @@ func NewServergRPC(addr string, group []*Config) (*ServergRPC, error) { tlsConfig = conf.TLSConfig } - return &ServergRPC{Server: s, tlsConfig: tlsConfig, watch: watch.NewWatcher(watchables(s.zones))}, nil + return &ServergRPC{Server: s, tlsConfig: tlsConfig}, nil } // Serve implements caddy.TCPServer interface. @@ -103,9 +101,6 @@ func (s *ServergRPC) OnStartupComplete() { func (s *ServergRPC) Stop() (err error) { s.m.Lock() defer s.m.Unlock() - if s.watch != nil { - s.watch.Stop() - } if s.grpcServer != nil { s.grpcServer.GracefulStop() } @@ -144,12 +139,6 @@ func (s *ServergRPC) Query(ctx context.Context, in *pb.DnsPacket) (*pb.DnsPacket return &pb.DnsPacket{Msg: packed}, nil } -// Watch is the entrypoint called by the gRPC layer when the user asks -// to watch a query. -func (s *ServergRPC) Watch(stream pb.DnsService_WatchServer) error { - return s.watch.Watch(stream) -} - // Shutdown stops the server (non gracefully). func (s *ServergRPC) Shutdown() error { if s.grpcServer != nil { diff --git a/core/dnsserver/watch.go b/core/dnsserver/watch.go deleted file mode 100644 index 590bac144..000000000 --- a/core/dnsserver/watch.go +++ /dev/null @@ -1,18 +0,0 @@ -package dnsserver - -import ( - "github.com/coredns/coredns/plugin/pkg/watch" -) - -func watchables(zones map[string]*Config) []watch.Watchable { - var w []watch.Watchable - for _, config := range zones { - plugins := config.Handlers() - for _, p := range plugins { - if x, ok := p.(watch.Watchable); ok { - w = append(w, x) - } - } - } - return w -} diff --git a/pb/Makefile b/pb/Makefile index bdfc8e3b0..a666d6c15 100644 --- a/pb/Makefile +++ b/pb/Makefile @@ -6,6 +6,8 @@ all: dns.pb.go dns.pb.go: dns.proto - protoc --go_out=plugins=grpc:. dns.proto && \ - sed -e s?golang.org/x/net/context?context? < dns.pb.go > dns.pb.go.tmp && \ - mv dns.pb.go.tmp dns.pb.go + protoc --go_out=plugins=grpc:. dns.proto + +.PHONY: clean +clean: + rm dns.pb.go diff --git a/pb/dns.pb.go b/pb/dns.pb.go index d79e24f6d..919f0e7d7 100644 --- a/pb/dns.pb.go +++ b/pb/dns.pb.go @@ -1,27 +1,14 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // source: dns.proto -/* -Package pb is a generated protocol buffer package. - -It is generated from these files: - dns.proto - -It has these top-level messages: - DnsPacket - WatchRequest - WatchCreateRequest - WatchCancelRequest - WatchResponse -*/ package pb -import proto "github.com/golang/protobuf/proto" -import fmt "fmt" -import math "math" - import ( context "context" + fmt "fmt" + math "math" + + proto "github.com/golang/protobuf/proto" grpc "google.golang.org/grpc" ) @@ -30,20 +17,46 @@ var _ = proto.Marshal var _ = fmt.Errorf var _ = math.Inf +/* Miek: disabled this manually, because I don't know what the heck */ +/* // 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.ProtoPackageIsVersion2 // please upgrade the proto package +const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package +*/ type DnsPacket struct { - Msg []byte `protobuf:"bytes,1,opt,name=msg,proto3" json:"msg,omitempty"` + Msg []byte `protobuf:"bytes,1,opt,name=msg,proto3" json:"msg,omitempty"` + XXX_NoUnkeyedLiteral struct{} `json:"-"` + XXX_unrecognized []byte `json:"-"` + XXX_sizecache int32 `json:"-"` } -func (m *DnsPacket) Reset() { *m = DnsPacket{} } -func (m *DnsPacket) String() string { return proto.CompactTextString(m) } -func (*DnsPacket) ProtoMessage() {} -func (*DnsPacket) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} } +func (m *DnsPacket) Reset() { *m = DnsPacket{} } +func (m *DnsPacket) String() string { return proto.CompactTextString(m) } +func (*DnsPacket) ProtoMessage() {} +func (*DnsPacket) Descriptor() ([]byte, []int) { + return fileDescriptor_638ff8d8aaf3d8ae, []int{0} +} + +func (m *DnsPacket) XXX_Unmarshal(b []byte) error { + return xxx_messageInfo_DnsPacket.Unmarshal(m, b) +} +func (m *DnsPacket) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + return xxx_messageInfo_DnsPacket.Marshal(b, m, deterministic) +} +func (m *DnsPacket) XXX_Merge(src proto.Message) { + xxx_messageInfo_DnsPacket.Merge(m, src) +} +func (m *DnsPacket) XXX_Size() int { + return xxx_messageInfo_DnsPacket.Size(m) +} +func (m *DnsPacket) XXX_DiscardUnknown() { + xxx_messageInfo_DnsPacket.DiscardUnknown(m) +} + +var xxx_messageInfo_DnsPacket proto.InternalMessageInfo func (m *DnsPacket) GetMsg() []byte { if m != nil { @@ -52,223 +65,22 @@ func (m *DnsPacket) GetMsg() []byte { return nil } -type WatchRequest struct { - // request_union is a request to either create a new watcher or cancel an existing watcher. - // - // Types that are valid to be assigned to RequestUnion: - // *WatchRequest_CreateRequest - // *WatchRequest_CancelRequest - RequestUnion isWatchRequest_RequestUnion `protobuf_oneof:"request_union"` -} - -func (m *WatchRequest) Reset() { *m = WatchRequest{} } -func (m *WatchRequest) String() string { return proto.CompactTextString(m) } -func (*WatchRequest) ProtoMessage() {} -func (*WatchRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{1} } - -type isWatchRequest_RequestUnion interface { - isWatchRequest_RequestUnion() -} - -type WatchRequest_CreateRequest struct { - CreateRequest *WatchCreateRequest `protobuf:"bytes,1,opt,name=create_request,json=createRequest,oneof"` -} -type WatchRequest_CancelRequest struct { - CancelRequest *WatchCancelRequest `protobuf:"bytes,2,opt,name=cancel_request,json=cancelRequest,oneof"` -} - -func (*WatchRequest_CreateRequest) isWatchRequest_RequestUnion() {} -func (*WatchRequest_CancelRequest) isWatchRequest_RequestUnion() {} - -func (m *WatchRequest) GetRequestUnion() isWatchRequest_RequestUnion { - if m != nil { - return m.RequestUnion - } - return nil -} - -func (m *WatchRequest) GetCreateRequest() *WatchCreateRequest { - if x, ok := m.GetRequestUnion().(*WatchRequest_CreateRequest); ok { - return x.CreateRequest - } - return nil -} - -func (m *WatchRequest) GetCancelRequest() *WatchCancelRequest { - if x, ok := m.GetRequestUnion().(*WatchRequest_CancelRequest); ok { - return x.CancelRequest - } - return nil -} - -// XXX_OneofFuncs is for the internal use of the proto package. -func (*WatchRequest) XXX_OneofFuncs() (func(msg proto.Message, b *proto.Buffer) error, func(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error), func(msg proto.Message) (n int), []interface{}) { - return _WatchRequest_OneofMarshaler, _WatchRequest_OneofUnmarshaler, _WatchRequest_OneofSizer, []interface{}{ - (*WatchRequest_CreateRequest)(nil), - (*WatchRequest_CancelRequest)(nil), - } -} - -func _WatchRequest_OneofMarshaler(msg proto.Message, b *proto.Buffer) error { - m := msg.(*WatchRequest) - // request_union - switch x := m.RequestUnion.(type) { - case *WatchRequest_CreateRequest: - b.EncodeVarint(1<<3 | proto.WireBytes) - if err := b.EncodeMessage(x.CreateRequest); err != nil { - return err - } - case *WatchRequest_CancelRequest: - b.EncodeVarint(2<<3 | proto.WireBytes) - if err := b.EncodeMessage(x.CancelRequest); err != nil { - return err - } - case nil: - default: - return fmt.Errorf("WatchRequest.RequestUnion has unexpected type %T", x) - } - return nil -} - -func _WatchRequest_OneofUnmarshaler(msg proto.Message, tag, wire int, b *proto.Buffer) (bool, error) { - m := msg.(*WatchRequest) - switch tag { - case 1: // request_union.create_request - if wire != proto.WireBytes { - return true, proto.ErrInternalBadWireType - } - msg := new(WatchCreateRequest) - err := b.DecodeMessage(msg) - m.RequestUnion = &WatchRequest_CreateRequest{msg} - return true, err - case 2: // request_union.cancel_request - if wire != proto.WireBytes { - return true, proto.ErrInternalBadWireType - } - msg := new(WatchCancelRequest) - err := b.DecodeMessage(msg) - m.RequestUnion = &WatchRequest_CancelRequest{msg} - return true, err - default: - return false, nil - } -} - -func _WatchRequest_OneofSizer(msg proto.Message) (n int) { - m := msg.(*WatchRequest) - // request_union - switch x := m.RequestUnion.(type) { - case *WatchRequest_CreateRequest: - s := proto.Size(x.CreateRequest) - n += proto.SizeVarint(1<<3 | proto.WireBytes) - n += proto.SizeVarint(uint64(s)) - n += s - case *WatchRequest_CancelRequest: - s := proto.Size(x.CancelRequest) - n += proto.SizeVarint(2<<3 | proto.WireBytes) - n += proto.SizeVarint(uint64(s)) - n += s - case nil: - default: - panic(fmt.Sprintf("proto: unexpected type %T in oneof", x)) - } - return n -} - -type WatchCreateRequest struct { - Query *DnsPacket `protobuf:"bytes,1,opt,name=query" json:"query,omitempty"` -} - -func (m *WatchCreateRequest) Reset() { *m = WatchCreateRequest{} } -func (m *WatchCreateRequest) String() string { return proto.CompactTextString(m) } -func (*WatchCreateRequest) ProtoMessage() {} -func (*WatchCreateRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{2} } - -func (m *WatchCreateRequest) GetQuery() *DnsPacket { - if m != nil { - return m.Query - } - return nil -} - -type WatchCancelRequest struct { - // watch_id is the watcher id to cancel - WatchId int64 `protobuf:"varint,1,opt,name=watch_id,json=watchId" json:"watch_id,omitempty"` -} - -func (m *WatchCancelRequest) Reset() { *m = WatchCancelRequest{} } -func (m *WatchCancelRequest) String() string { return proto.CompactTextString(m) } -func (*WatchCancelRequest) ProtoMessage() {} -func (*WatchCancelRequest) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{3} } - -func (m *WatchCancelRequest) GetWatchId() int64 { - if m != nil { - return m.WatchId - } - return 0 -} - -type WatchResponse struct { - // watch_id is the ID of the watcher that corresponds to the response. - WatchId int64 `protobuf:"varint,1,opt,name=watch_id,json=watchId" json:"watch_id,omitempty"` - // created is set to true if the response is for a create watch request. - // The client should record the watch_id and expect to receive DNS replies - // from the same stream. - // All replies sent to the created watcher will attach with the same watch_id. - Created bool `protobuf:"varint,2,opt,name=created" json:"created,omitempty"` - // canceled is set to true if the response is for a cancel watch request. - // No further events will be sent to the canceled watcher. - Canceled bool `protobuf:"varint,3,opt,name=canceled" json:"canceled,omitempty"` - Qname string `protobuf:"bytes,4,opt,name=qname" json:"qname,omitempty"` - Err string `protobuf:"bytes,5,opt,name=err" json:"err,omitempty"` -} - -func (m *WatchResponse) Reset() { *m = WatchResponse{} } -func (m *WatchResponse) String() string { return proto.CompactTextString(m) } -func (*WatchResponse) ProtoMessage() {} -func (*WatchResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{4} } - -func (m *WatchResponse) GetWatchId() int64 { - if m != nil { - return m.WatchId - } - return 0 -} - -func (m *WatchResponse) GetCreated() bool { - if m != nil { - return m.Created - } - return false -} - -func (m *WatchResponse) GetCanceled() bool { - if m != nil { - return m.Canceled - } - return false -} - -func (m *WatchResponse) GetQname() string { - if m != nil { - return m.Qname - } - return "" -} - -func (m *WatchResponse) GetErr() string { - if m != nil { - return m.Err - } - return "" -} - func init() { proto.RegisterType((*DnsPacket)(nil), "coredns.dns.DnsPacket") - proto.RegisterType((*WatchRequest)(nil), "coredns.dns.WatchRequest") - proto.RegisterType((*WatchCreateRequest)(nil), "coredns.dns.WatchCreateRequest") - proto.RegisterType((*WatchCancelRequest)(nil), "coredns.dns.WatchCancelRequest") - proto.RegisterType((*WatchResponse)(nil), "coredns.dns.WatchResponse") +} + +func init() { proto.RegisterFile("dns.proto", fileDescriptor_638ff8d8aaf3d8ae) } + +var fileDescriptor_638ff8d8aaf3d8ae = []byte{ + // 120 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4c, 0xc9, 0x2b, 0xd6, + 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0xe2, 0x4e, 0xce, 0x2f, 0x4a, 0x05, 0x71, 0x53, 0xf2, 0x8a, + 0x95, 0x64, 0xb9, 0x38, 0x5d, 0xf2, 0x8a, 0x03, 0x12, 0x93, 0xb3, 0x53, 0x4b, 0x84, 0x04, 0xb8, + 0x98, 0x73, 0x8b, 0xd3, 0x25, 0x18, 0x15, 0x18, 0x35, 0x78, 0x82, 0x40, 0x4c, 0x23, 0x57, 0x2e, + 0x2e, 0x97, 0xbc, 0xe2, 0xe0, 0xd4, 0xa2, 0xb2, 0xcc, 0xe4, 0x54, 0x21, 0x73, 0x2e, 0xd6, 0xc0, + 0xd2, 0xd4, 0xa2, 0x4a, 0x21, 0x31, 0x3d, 0x24, 0x33, 0xf4, 0xe0, 0x06, 0x48, 0xe1, 0x10, 0x77, + 0x62, 0x89, 0x62, 0x2a, 0x48, 0x4a, 0x62, 0x03, 0xdb, 0x6f, 0x0c, 0x08, 0x00, 0x00, 0xff, 0xff, + 0xf5, 0xd1, 0x3f, 0x26, 0x8c, 0x00, 0x00, 0x00, } // Reference imports to suppress errors if they are not otherwise used. @@ -279,11 +91,11 @@ var _ grpc.ClientConn // is compatible with the grpc package it is being compiled against. const _ = grpc.SupportPackageIsVersion4 -// Client API for DnsService service - +// DnsServiceClient is the client API for DnsService service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream. type DnsServiceClient interface { Query(ctx context.Context, in *DnsPacket, opts ...grpc.CallOption) (*DnsPacket, error) - Watch(ctx context.Context, opts ...grpc.CallOption) (DnsService_WatchClient, error) } type dnsServiceClient struct { @@ -296,49 +108,16 @@ func NewDnsServiceClient(cc *grpc.ClientConn) DnsServiceClient { func (c *dnsServiceClient) Query(ctx context.Context, in *DnsPacket, opts ...grpc.CallOption) (*DnsPacket, error) { out := new(DnsPacket) - err := grpc.Invoke(ctx, "/coredns.dns.DnsService/Query", in, out, c.cc, opts...) + err := c.cc.Invoke(ctx, "/coredns.dns.DnsService/Query", in, out, opts...) if err != nil { return nil, err } return out, nil } -func (c *dnsServiceClient) Watch(ctx context.Context, opts ...grpc.CallOption) (DnsService_WatchClient, error) { - stream, err := grpc.NewClientStream(ctx, &_DnsService_serviceDesc.Streams[0], c.cc, "/coredns.dns.DnsService/Watch", opts...) - if err != nil { - return nil, err - } - x := &dnsServiceWatchClient{stream} - return x, nil -} - -type DnsService_WatchClient interface { - Send(*WatchRequest) error - Recv() (*WatchResponse, error) - grpc.ClientStream -} - -type dnsServiceWatchClient struct { - grpc.ClientStream -} - -func (x *dnsServiceWatchClient) Send(m *WatchRequest) error { - return x.ClientStream.SendMsg(m) -} - -func (x *dnsServiceWatchClient) Recv() (*WatchResponse, error) { - m := new(WatchResponse) - if err := x.ClientStream.RecvMsg(m); err != nil { - return nil, err - } - return m, nil -} - -// Server API for DnsService service - +// DnsServiceServer is the server API for DnsService service. type DnsServiceServer interface { Query(context.Context, *DnsPacket) (*DnsPacket, error) - Watch(DnsService_WatchServer) error } func RegisterDnsServiceServer(s *grpc.Server, srv DnsServiceServer) { @@ -363,32 +142,6 @@ func _DnsService_Query_Handler(srv interface{}, ctx context.Context, dec func(in return interceptor(ctx, in, info, handler) } -func _DnsService_Watch_Handler(srv interface{}, stream grpc.ServerStream) error { - return srv.(DnsServiceServer).Watch(&dnsServiceWatchServer{stream}) -} - -type DnsService_WatchServer interface { - Send(*WatchResponse) error - Recv() (*WatchRequest, error) - grpc.ServerStream -} - -type dnsServiceWatchServer struct { - grpc.ServerStream -} - -func (x *dnsServiceWatchServer) Send(m *WatchResponse) error { - return x.ServerStream.SendMsg(m) -} - -func (x *dnsServiceWatchServer) Recv() (*WatchRequest, error) { - m := new(WatchRequest) - if err := x.ServerStream.RecvMsg(m); err != nil { - return nil, err - } - return m, nil -} - var _DnsService_serviceDesc = grpc.ServiceDesc{ ServiceName: "coredns.dns.DnsService", HandlerType: (*DnsServiceServer)(nil), @@ -398,40 +151,6 @@ var _DnsService_serviceDesc = grpc.ServiceDesc{ Handler: _DnsService_Query_Handler, }, }, - Streams: []grpc.StreamDesc{ - { - StreamName: "Watch", - Handler: _DnsService_Watch_Handler, - ServerStreams: true, - ClientStreams: true, - }, - }, + Streams: []grpc.StreamDesc{}, Metadata: "dns.proto", } - -func init() { proto.RegisterFile("dns.proto", fileDescriptor0) } - -var fileDescriptor0 = []byte{ - // 333 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x7c, 0x52, 0x41, 0x4f, 0x32, 0x31, - 0x14, 0xfc, 0x0a, 0xec, 0x07, 0x3c, 0x40, 0xcd, 0x8b, 0x31, 0xcb, 0x26, 0x46, 0xb2, 0x27, 0x0e, - 0x06, 0x0d, 0x1e, 0xbc, 0xaf, 0x1c, 0xf0, 0xa6, 0xf5, 0x60, 0xe2, 0x85, 0x2c, 0xdd, 0x17, 0x25, - 0x4a, 0x17, 0xda, 0x45, 0xe3, 0x3f, 0xd0, 0xdf, 0xe3, 0x1f, 0x34, 0xfb, 0xba, 0x90, 0x35, 0x88, - 0xb7, 0xce, 0x74, 0x3a, 0xed, 0xcc, 0x2b, 0x34, 0x13, 0x6d, 0x07, 0x0b, 0x93, 0x66, 0x29, 0xb6, - 0x54, 0x6a, 0x28, 0x87, 0x89, 0xb6, 0xe1, 0x31, 0x34, 0x47, 0xda, 0xde, 0xc4, 0xea, 0x99, 0x32, - 0x3c, 0x80, 0xea, 0xdc, 0x3e, 0xfa, 0xa2, 0x27, 0xfa, 0x6d, 0x99, 0x2f, 0xc3, 0x2f, 0x01, 0xed, - 0xfb, 0x38, 0x53, 0x4f, 0x92, 0x96, 0x2b, 0xb2, 0x19, 0x8e, 0x61, 0x4f, 0x19, 0x8a, 0x33, 0x9a, - 0x18, 0xc7, 0xb0, 0xba, 0x35, 0x3c, 0x19, 0x94, 0x5c, 0x07, 0x7c, 0xe4, 0x8a, 0x75, 0xc5, 0xc1, - 0xf1, 0x3f, 0xd9, 0x51, 0x65, 0x82, 0x9d, 0x62, 0xad, 0xe8, 0x65, 0xe3, 0x54, 0xd9, 0xe9, 0xc4, - 0xba, 0xb2, 0x53, 0x99, 0x88, 0xf6, 0xa1, 0x53, 0x58, 0x4c, 0x56, 0x7a, 0x96, 0xea, 0x30, 0x02, - 0xdc, 0x7e, 0x01, 0x9e, 0x82, 0xb7, 0x5c, 0x91, 0x79, 0x2f, 0x5e, 0x7c, 0xf4, 0xe3, 0x9e, 0x4d, - 0x09, 0xd2, 0x89, 0xc2, 0xb3, 0xb5, 0x47, 0xf9, 0x2a, 0xec, 0x42, 0xe3, 0x2d, 0x67, 0x27, 0xb3, - 0x84, 0x6d, 0xaa, 0xb2, 0xce, 0xf8, 0x3a, 0x09, 0x3f, 0x04, 0x74, 0x8a, 0xaa, 0xec, 0x22, 0xd5, - 0x96, 0xfe, 0x10, 0xa3, 0x0f, 0x75, 0xd7, 0x46, 0xc2, 0xa9, 0x1b, 0x72, 0x0d, 0x31, 0x80, 0x86, - 0x4b, 0x47, 0x89, 0x5f, 0xe5, 0xad, 0x0d, 0xc6, 0x43, 0xf0, 0x96, 0x3a, 0x9e, 0x93, 0x5f, 0xeb, - 0x89, 0x7e, 0x53, 0x3a, 0x90, 0x4f, 0x8d, 0x8c, 0xf1, 0x3d, 0xe6, 0xf2, 0xe5, 0xf0, 0x53, 0x00, - 0x8c, 0xb4, 0xbd, 0x23, 0xf3, 0x3a, 0x53, 0x84, 0x97, 0xe0, 0xdd, 0xe6, 0x99, 0x70, 0x47, 0xe4, - 0x60, 0x07, 0x8f, 0x11, 0x78, 0x9c, 0x08, 0xbb, 0xdb, 0x33, 0x29, 0x1a, 0x09, 0x82, 0xdf, 0xb6, - 0x5c, 0x01, 0x7d, 0x71, 0x2e, 0xa2, 0xda, 0x43, 0x65, 0x31, 0x9d, 0xfe, 0xe7, 0xaf, 0x77, 0xf1, - 0x1d, 0x00, 0x00, 0xff, 0xff, 0xd2, 0x5b, 0x8c, 0xe1, 0x87, 0x02, 0x00, 0x00, -} diff --git a/pb/dns.proto b/pb/dns.proto index e4ac2eb2c..8461f01e6 100644 --- a/pb/dns.proto +++ b/pb/dns.proto @@ -9,41 +9,4 @@ message DnsPacket { service DnsService { rpc Query (DnsPacket) returns (DnsPacket); - rpc Watch (stream WatchRequest) returns (stream WatchResponse); -} - -message WatchRequest { - // request_union is a request to either create a new watcher or cancel an existing watcher. - oneof request_union { - WatchCreateRequest create_request = 1; - WatchCancelRequest cancel_request = 2; - } -} - -message WatchCreateRequest { - DnsPacket query = 1; -} - -message WatchCancelRequest { - // watch_id is the watcher id to cancel - int64 watch_id = 1; -} - -message WatchResponse { - // watch_id is the ID of the watcher that corresponds to the response. - int64 watch_id = 1; - - // created is set to true if the response is for a create watch request. - // The client should record the watch_id and expect to receive DNS replies - // from the same stream. - // All replies sent to the created watcher will attach with the same watch_id. - bool created = 2; - - // canceled is set to true if the response is for a cancel watch request. - // No further events will be sent to the canceled watcher. - bool canceled = 3; - - string qname = 4; - - string err = 5; } diff --git a/plugin/federation/kubernetes_api_test.go b/plugin/federation/kubernetes_api_test.go index 4b62605d1..35b058b07 100644 --- a/plugin/federation/kubernetes_api_test.go +++ b/plugin/federation/kubernetes_api_test.go @@ -3,7 +3,6 @@ package federation import ( "github.com/coredns/coredns/plugin/kubernetes" "github.com/coredns/coredns/plugin/kubernetes/object" - "github.com/coredns/coredns/plugin/pkg/watch" api "k8s.io/api/core/v1" meta "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -19,9 +18,6 @@ func (APIConnFederationTest) Stop() error { return func (APIConnFederationTest) SvcIndexReverse(string) []*object.Service { return nil } func (APIConnFederationTest) EpIndexReverse(string) []*object.Endpoints { return nil } func (APIConnFederationTest) Modified() int64 { return 0 } -func (APIConnFederationTest) SetWatchChan(watch.Chan) {} -func (APIConnFederationTest) Watch(string) error { return nil } -func (APIConnFederationTest) StopWatching(string) {} func (APIConnFederationTest) PodIndex(string) []*object.Pod { return []*object.Pod{ diff --git a/plugin/k8s_external/external_test.go b/plugin/k8s_external/external_test.go index a384cedff..621ff9933 100644 --- a/plugin/k8s_external/external_test.go +++ b/plugin/k8s_external/external_test.go @@ -7,7 +7,6 @@ import ( "github.com/coredns/coredns/plugin/kubernetes" "github.com/coredns/coredns/plugin/kubernetes/object" "github.com/coredns/coredns/plugin/pkg/dnstest" - "github.com/coredns/coredns/plugin/pkg/watch" "github.com/coredns/coredns/plugin/test" "github.com/coredns/coredns/request" @@ -158,9 +157,6 @@ func (external) Stop() error { return nil } func (external) EpIndexReverse(string) []*object.Endpoints { return nil } func (external) SvcIndexReverse(string) []*object.Service { return nil } func (external) Modified() int64 { return 0 } -func (external) SetWatchChan(watch.Chan) {} -func (external) Watch(string) error { return nil } -func (external) StopWatching(string) {} func (external) EpIndex(s string) []*object.Endpoints { return nil } func (external) EndpointsList() []*object.Endpoints { return nil } func (external) GetNodeByName(name string) (*api.Node, error) { return nil, nil } diff --git a/plugin/kubernetes/README.md b/plugin/kubernetes/README.md index 464c04990..77add2395 100644 --- a/plugin/kubernetes/README.md +++ b/plugin/kubernetes/README.md @@ -112,11 +112,6 @@ kubernetes [ZONES...] { This plugin implements dynamic health checking. Currently this is limited to reporting healthy when the API has synced. -## Watch - -This plugin implements watch. A client that connects to CoreDNS using `coredns/client` can be notified -of changes to A, AAAA, and SRV records for Kubernetes services and endpoints. - ## Examples Handle all queries in the `cluster.local` zone. Connect to Kubernetes in-cluster. Also handle all diff --git a/plugin/kubernetes/controller.go b/plugin/kubernetes/controller.go index 9a2e9994a..92cb3758a 100644 --- a/plugin/kubernetes/controller.go +++ b/plugin/kubernetes/controller.go @@ -8,13 +8,11 @@ import ( "time" "github.com/coredns/coredns/plugin/kubernetes/object" - dnswatch "github.com/coredns/coredns/plugin/pkg/watch" api "k8s.io/api/core/v1" meta "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/watch" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" ) @@ -45,11 +43,6 @@ type dnsController interface { // Modified returns the timestamp of the most recent changes Modified() int64 - - // Watch-related items - SetWatchChan(dnswatch.Chan) - Watch(string) error - StopWatching(string) } type dnsControl struct { @@ -79,9 +72,6 @@ type dnsControl struct { shutdown bool stopCh chan struct{} - // watch-related items channel - watchChan dnswatch.Chan - watched map[string]struct{} zones []string endpointNameMode bool } @@ -105,7 +95,6 @@ func newdnsController(kubeClient kubernetes.Interface, opts dnsControlOpts) *dns client: kubeClient, selector: opts.selector, stopCh: make(chan struct{}), - watched: make(map[string]struct{}), zones: opts.zones, endpointNameMode: opts.endpointNameMode, } @@ -117,7 +106,7 @@ func newdnsController(kubeClient kubernetes.Interface, opts dnsControlOpts) *dns }, &api.Service{}, opts.resyncPeriod, - cache.ResourceEventHandlerFuncs{AddFunc: dns.Add, UpdateFunc: dns.Update, DeleteFunc: dns.Delete}, + cache.ResourceEventHandlerFuncs{}, cache.Indexers{svcNameNamespaceIndex: svcNameNamespaceIndexFunc, svcIPIndex: svcIPIndexFunc}, object.ToService, ) @@ -130,7 +119,7 @@ func newdnsController(kubeClient kubernetes.Interface, opts dnsControlOpts) *dns }, &api.Pod{}, opts.resyncPeriod, - cache.ResourceEventHandlerFuncs{AddFunc: dns.Add, UpdateFunc: dns.Update, DeleteFunc: dns.Delete}, + cache.ResourceEventHandlerFuncs{}, cache.Indexers{podIPIndex: podIPIndexFunc}, object.ToPod, ) @@ -144,7 +133,7 @@ func newdnsController(kubeClient kubernetes.Interface, opts dnsControlOpts) *dns }, &api.Endpoints{}, opts.resyncPeriod, - cache.ResourceEventHandlerFuncs{AddFunc: dns.Add, UpdateFunc: dns.Update, DeleteFunc: dns.Delete}, + cache.ResourceEventHandlerFuncs{}, cache.Indexers{epNameNamespaceIndex: epNameNamespaceIndexFunc, epIPIndex: epIPIndexFunc}, object.ToEndpoints) } @@ -223,26 +212,6 @@ func podListFunc(c kubernetes.Interface, ns string, s labels.Selector) func(meta } } -func serviceWatchFunc(c kubernetes.Interface, ns string, s labels.Selector) func(options meta.ListOptions) (watch.Interface, error) { - return func(options meta.ListOptions) (watch.Interface, error) { - if s != nil { - options.LabelSelector = s.String() - } - w, err := c.CoreV1().Services(ns).Watch(options) - return w, err - } -} - -func podWatchFunc(c kubernetes.Interface, ns string, s labels.Selector) func(options meta.ListOptions) (watch.Interface, error) { - return func(options meta.ListOptions) (watch.Interface, error) { - if s != nil { - options.LabelSelector = s.String() - } - w, err := c.CoreV1().Pods(ns).Watch(options) - return w, err - } -} - func endpointsListFunc(c kubernetes.Interface, ns string, s labels.Selector) func(meta.ListOptions) (runtime.Object, error) { return func(opts meta.ListOptions) (runtime.Object, error) { if s != nil { @@ -253,16 +222,6 @@ func endpointsListFunc(c kubernetes.Interface, ns string, s labels.Selector) fun } } -func endpointsWatchFunc(c kubernetes.Interface, ns string, s labels.Selector) func(options meta.ListOptions) (watch.Interface, error) { - return func(options meta.ListOptions) (watch.Interface, error) { - if s != nil { - options.LabelSelector = s.String() - } - w, err := c.CoreV1().Endpoints(ns).Watch(options) - return w, err - } -} - func namespaceListFunc(c kubernetes.Interface, s labels.Selector) func(meta.ListOptions) (runtime.Object, error) { return func(opts meta.ListOptions) (runtime.Object, error) { if s != nil { @@ -273,27 +232,6 @@ func namespaceListFunc(c kubernetes.Interface, s labels.Selector) func(meta.List } } -func namespaceWatchFunc(c kubernetes.Interface, s labels.Selector) func(options meta.ListOptions) (watch.Interface, error) { - return func(options meta.ListOptions) (watch.Interface, error) { - if s != nil { - options.LabelSelector = s.String() - } - w, err := c.CoreV1().Namespaces().Watch(options) - return w, err - } -} - -func (dns *dnsControl) SetWatchChan(c dnswatch.Chan) { dns.watchChan = c } -func (dns *dnsControl) StopWatching(qname string) { delete(dns.watched, qname) } - -func (dns *dnsControl) Watch(qname string) error { - if dns.watchChan == nil { - return fmt.Errorf("cannot start watch because the channel has not been set") - } - dns.watched[qname] = struct{}{} - return nil -} - // Stop stops the controller. func (dns *dnsControl) Stop() error { dns.stopLock.Lock() diff --git a/plugin/kubernetes/external_test.go b/plugin/kubernetes/external_test.go index 3636a176e..d3762961a 100644 --- a/plugin/kubernetes/external_test.go +++ b/plugin/kubernetes/external_test.go @@ -5,7 +5,6 @@ import ( "github.com/coredns/coredns/plugin/etcd/msg" "github.com/coredns/coredns/plugin/kubernetes/object" - "github.com/coredns/coredns/plugin/pkg/watch" "github.com/coredns/coredns/plugin/test" "github.com/coredns/coredns/request" @@ -85,9 +84,6 @@ func (external) Stop() error { return nil } func (external) EpIndexReverse(string) []*object.Endpoints { return nil } func (external) SvcIndexReverse(string) []*object.Service { return nil } func (external) Modified() int64 { return 0 } -func (external) SetWatchChan(watch.Chan) {} -func (external) Watch(string) error { return nil } -func (external) StopWatching(string) {} func (external) EpIndex(s string) []*object.Endpoints { return nil } func (external) EndpointsList() []*object.Endpoints { return nil } func (external) GetNodeByName(name string) (*api.Node, error) { return nil, nil } diff --git a/plugin/kubernetes/handler_test.go b/plugin/kubernetes/handler_test.go index 6b0724045..8dbacbd14 100644 --- a/plugin/kubernetes/handler_test.go +++ b/plugin/kubernetes/handler_test.go @@ -7,7 +7,6 @@ import ( "github.com/coredns/coredns/plugin/kubernetes/object" "github.com/coredns/coredns/plugin/pkg/dnstest" - "github.com/coredns/coredns/plugin/pkg/watch" "github.com/coredns/coredns/plugin/test" "github.com/miekg/dns" @@ -441,9 +440,6 @@ func (APIConnServeTest) Stop() error { return nil func (APIConnServeTest) EpIndexReverse(string) []*object.Endpoints { return nil } func (APIConnServeTest) SvcIndexReverse(string) []*object.Service { return nil } func (APIConnServeTest) Modified() int64 { return time.Now().Unix() } -func (APIConnServeTest) SetWatchChan(watch.Chan) {} -func (APIConnServeTest) Watch(string) error { return nil } -func (APIConnServeTest) StopWatching(string) {} func (APIConnServeTest) PodIndex(string) []*object.Pod { a := []*object.Pod{ diff --git a/plugin/kubernetes/kubernetes_test.go b/plugin/kubernetes/kubernetes_test.go index 3fe135694..6b661f7d5 100644 --- a/plugin/kubernetes/kubernetes_test.go +++ b/plugin/kubernetes/kubernetes_test.go @@ -5,7 +5,6 @@ import ( "github.com/coredns/coredns/plugin" "github.com/coredns/coredns/plugin/kubernetes/object" - "github.com/coredns/coredns/plugin/pkg/watch" "github.com/coredns/coredns/request" "github.com/miekg/dns" @@ -66,9 +65,6 @@ func (APIConnServiceTest) PodIndex(string) []*object.Pod { return ni func (APIConnServiceTest) SvcIndexReverse(string) []*object.Service { return nil } func (APIConnServiceTest) EpIndexReverse(string) []*object.Endpoints { return nil } func (APIConnServiceTest) Modified() int64 { return 0 } -func (APIConnServiceTest) SetWatchChan(watch.Chan) {} -func (APIConnServiceTest) Watch(string) error { return nil } -func (APIConnServiceTest) StopWatching(string) {} func (APIConnServiceTest) SvcIndex(string) []*object.Service { svcs := []*object.Service{ diff --git a/plugin/kubernetes/ns_test.go b/plugin/kubernetes/ns_test.go index fd781bc14..df7cf5f83 100644 --- a/plugin/kubernetes/ns_test.go +++ b/plugin/kubernetes/ns_test.go @@ -4,7 +4,6 @@ import ( "testing" "github.com/coredns/coredns/plugin/kubernetes/object" - "github.com/coredns/coredns/plugin/pkg/watch" api "k8s.io/api/core/v1" ) @@ -20,9 +19,6 @@ func (APIConnTest) SvcIndexReverse(string) []*object.Service { return nil } func (APIConnTest) EpIndex(string) []*object.Endpoints { return nil } func (APIConnTest) EndpointsList() []*object.Endpoints { return nil } func (APIConnTest) Modified() int64 { return 0 } -func (APIConnTest) SetWatchChan(watch.Chan) {} -func (APIConnTest) Watch(string) error { return nil } -func (APIConnTest) StopWatching(string) {} func (APIConnTest) ServiceList() []*object.Service { svcs := []*object.Service{ diff --git a/plugin/kubernetes/reverse_test.go b/plugin/kubernetes/reverse_test.go index baaff4787..3c88fd4d5 100644 --- a/plugin/kubernetes/reverse_test.go +++ b/plugin/kubernetes/reverse_test.go @@ -6,7 +6,6 @@ import ( "github.com/coredns/coredns/plugin/kubernetes/object" "github.com/coredns/coredns/plugin/pkg/dnstest" - "github.com/coredns/coredns/plugin/pkg/watch" "github.com/coredns/coredns/plugin/test" "github.com/miekg/dns" @@ -24,9 +23,6 @@ func (APIConnReverseTest) EpIndex(string) []*object.Endpoints { return nil } func (APIConnReverseTest) EndpointsList() []*object.Endpoints { return nil } func (APIConnReverseTest) ServiceList() []*object.Service { return nil } func (APIConnReverseTest) Modified() int64 { return 0 } -func (APIConnReverseTest) SetWatchChan(watch.Chan) {} -func (APIConnReverseTest) Watch(string) error { return nil } -func (APIConnReverseTest) StopWatching(string) {} func (APIConnReverseTest) SvcIndex(svc string) []*object.Service { if svc != "svc1.testns" { diff --git a/plugin/kubernetes/watch.go b/plugin/kubernetes/watch.go index 4db14ad3a..52529a6b0 100644 --- a/plugin/kubernetes/watch.go +++ b/plugin/kubernetes/watch.go @@ -1,194 +1,48 @@ package kubernetes import ( - "github.com/coredns/coredns/plugin/kubernetes/object" - "github.com/coredns/coredns/plugin/pkg/watch" meta "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/kubernetes" ) -// SetWatchChan implements watch.Watchable -func (k *Kubernetes) SetWatchChan(c watch.Chan) { - k.APIConn.SetWatchChan(c) -} - -// Watch is called when a watch is started for a name. -func (k *Kubernetes) Watch(qname string) error { - return k.APIConn.Watch(qname) -} - -// StopWatching is called when no more watches remain for a name -func (k *Kubernetes) StopWatching(qname string) { - k.APIConn.StopWatching(qname) -} - -var _ watch.Watchable = &Kubernetes{} - -func (dns *dnsControl) sendServiceUpdates(s *object.Service) { - for i := range dns.zones { - name := serviceFQDN(s, dns.zones[i]) - if _, ok := dns.watched[name]; ok { - dns.watchChan <- name +func serviceWatchFunc(c kubernetes.Interface, ns string, s labels.Selector) func(options meta.ListOptions) (watch.Interface, error) { + return func(options meta.ListOptions) (watch.Interface, error) { + if s != nil { + options.LabelSelector = s.String() } + w, err := c.CoreV1().Services(ns).Watch(options) + return w, err } } -func (dns *dnsControl) sendPodUpdates(p *object.Pod) { - for i := range dns.zones { - name := podFQDN(p, dns.zones[i]) - if _, ok := dns.watched[name]; ok { - dns.watchChan <- name +func podWatchFunc(c kubernetes.Interface, ns string, s labels.Selector) func(options meta.ListOptions) (watch.Interface, error) { + return func(options meta.ListOptions) (watch.Interface, error) { + if s != nil { + options.LabelSelector = s.String() } + w, err := c.CoreV1().Pods(ns).Watch(options) + return w, err } } -func (dns *dnsControl) sendEndpointsUpdates(ep *object.Endpoints) { - for _, zone := range dns.zones { - for _, name := range endpointFQDN(ep, zone, dns.endpointNameMode) { - if _, ok := dns.watched[name]; ok { - dns.watchChan <- name - } - } - name := serviceFQDN(ep, zone) - if _, ok := dns.watched[name]; ok { - dns.watchChan <- name +func endpointsWatchFunc(c kubernetes.Interface, ns string, s labels.Selector) func(options meta.ListOptions) (watch.Interface, error) { + return func(options meta.ListOptions) (watch.Interface, error) { + if s != nil { + options.LabelSelector = s.String() } + w, err := c.CoreV1().Endpoints(ns).Watch(options) + return w, err } } -// endpointsSubsetDiffs returns an Endpoints struct containing the Subsets that have changed between a and b. -// When we notify clients of changed endpoints we only want to notify them of endpoints that have changed. -// The Endpoints API object holds more than one endpoint, held in a list of Subsets. Each Subset refers to -// an endpoint. So, here we create a new Endpoints struct, and populate it with only the endpoints that have changed. -// This new Endpoints object is later used to generate the list of endpoint FQDNs to send to the client. -// This function computes this literally by combining the sets (in a and not in b) union (in b and not in a). -func endpointsSubsetDiffs(a, b *object.Endpoints) *object.Endpoints { - c := b.CopyWithoutSubsets() - - // In the following loop, the first iteration computes (in a but not in b). - // The second iteration then adds (in b but not in a) - // The end result is an Endpoints that only contains the subsets (endpoints) that are different between a and b. - for _, abba := range [][]*object.Endpoints{{a, b}, {b, a}} { - a := abba[0] - b := abba[1] - left: - for _, as := range a.Subsets { - for _, bs := range b.Subsets { - if subsetsEquivalent(as, bs) { - continue left - } - } - c.Subsets = append(c.Subsets, as) +func namespaceWatchFunc(c kubernetes.Interface, s labels.Selector) func(options meta.ListOptions) (watch.Interface, error) { + return func(options meta.ListOptions) (watch.Interface, error) { + if s != nil { + options.LabelSelector = s.String() } - } - return c -} - -// sendUpdates sends a notification to the server if a watch is enabled for the qname. -func (dns *dnsControl) sendUpdates(oldObj, newObj interface{}) { - // If both objects have the same resource version, they are identical. - if newObj != nil && oldObj != nil && (oldObj.(meta.Object).GetResourceVersion() == newObj.(meta.Object).GetResourceVersion()) { - return - } - obj := newObj - if obj == nil { - obj = oldObj - } - switch ob := obj.(type) { - case *object.Service: - dns.updateModifed() - if len(dns.watched) == 0 { - return - } - dns.sendServiceUpdates(ob) - case *object.Endpoints: - if newObj == nil || oldObj == nil { - dns.updateModifed() - if len(dns.watched) == 0 { - return - } - dns.sendEndpointsUpdates(ob) - return - } - p := oldObj.(*object.Endpoints) - // endpoint updates can come frequently, make sure it's a change we care about - if endpointsEquivalent(p, ob) { - return - } - dns.updateModifed() - if len(dns.watched) == 0 { - return - } - dns.sendEndpointsUpdates(endpointsSubsetDiffs(p, ob)) - case *object.Pod: - dns.updateModifed() - if len(dns.watched) == 0 { - return - } - dns.sendPodUpdates(ob) - default: - log.Warningf("Updates for %T not supported.", ob) + w, err := c.CoreV1().Namespaces().Watch(options) + return w, err } } - -func (dns *dnsControl) Add(obj interface{}) { dns.sendUpdates(nil, obj) } -func (dns *dnsControl) Delete(obj interface{}) { dns.sendUpdates(obj, nil) } -func (dns *dnsControl) Update(oldObj, newObj interface{}) { dns.sendUpdates(oldObj, newObj) } - -// subsetsEquivalent checks if two endpoint subsets are significantly equivalent -// I.e. that they have the same ready addresses, host names, ports (including protocol -// and service names for SRV) -func subsetsEquivalent(sa, sb object.EndpointSubset) bool { - if len(sa.Addresses) != len(sb.Addresses) { - return false - } - if len(sa.Ports) != len(sb.Ports) { - return false - } - - // in Addresses and Ports, we should be able to rely on - // these being sorted and able to be compared - // they are supposed to be in a canonical format - for addr, aaddr := range sa.Addresses { - baddr := sb.Addresses[addr] - if aaddr.IP != baddr.IP { - return false - } - if aaddr.Hostname != baddr.Hostname { - return false - } - } - - for port, aport := range sa.Ports { - bport := sb.Ports[port] - if aport.Name != bport.Name { - return false - } - if aport.Port != bport.Port { - return false - } - if aport.Protocol != bport.Protocol { - return false - } - } - return true -} - -// endpointsEquivalent checks if the update to an endpoint is something -// that matters to us or if they are effectively equivalent. -func endpointsEquivalent(a, b *object.Endpoints) bool { - - if len(a.Subsets) != len(b.Subsets) { - return false - } - - // we should be able to rely on - // these being sorted and able to be compared - // they are supposed to be in a canonical format - for i, sa := range a.Subsets { - sb := b.Subsets[i] - if !subsetsEquivalent(sa, sb) { - return false - } - } - return true -} diff --git a/plugin/kubernetes/watch_test.go b/plugin/kubernetes/watch_test.go deleted file mode 100644 index 08eda894b..000000000 --- a/plugin/kubernetes/watch_test.go +++ /dev/null @@ -1,53 +0,0 @@ -package kubernetes - -import ( - "strconv" - "strings" - "testing" - - "github.com/coredns/coredns/plugin/kubernetes/object" -) - -func endpointSubsets(addrs ...string) (eps []object.EndpointSubset) { - for _, ap := range addrs { - apa := strings.Split(ap, ":") - address := apa[0] - port, _ := strconv.Atoi(apa[1]) - eps = append(eps, object.EndpointSubset{Addresses: []object.EndpointAddress{{IP: address}}, Ports: []object.EndpointPort{{Port: int32(port)}}}) - } - return eps -} - -func TestEndpointsSubsetDiffs(t *testing.T) { - var tests = []struct { - a, b, expected object.Endpoints - }{ - { // From a->b: Nothing changes - object.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")}, - object.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")}, - object.Endpoints{}, - }, - { // From a->b: Everything goes away - object.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")}, - object.Endpoints{}, - object.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")}, - }, - { // From a->b: Everything is new - object.Endpoints{}, - object.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")}, - object.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")}, - }, - { // From a->b: One goes away, one is new - object.Endpoints{Subsets: endpointSubsets("10.0.0.2:8080")}, - object.Endpoints{Subsets: endpointSubsets("10.0.0.1:80")}, - object.Endpoints{Subsets: endpointSubsets("10.0.0.2:8080", "10.0.0.1:80")}, - }, - } - - for i, te := range tests { - got := endpointsSubsetDiffs(&te.a, &te.b) - if !endpointsEquivalent(got, &te.expected) { - t.Errorf("Expected '%v' for test %v, got '%v'.", te.expected, i, got) - } - } -} diff --git a/plugin/kubernetes/xfr_test.go b/plugin/kubernetes/xfr_test.go index db55aa8ed..1fb3e0b43 100644 --- a/plugin/kubernetes/xfr_test.go +++ b/plugin/kubernetes/xfr_test.go @@ -5,7 +5,6 @@ import ( "strings" "testing" - "github.com/coredns/coredns/plugin/kubernetes/object" "github.com/coredns/coredns/plugin/pkg/dnstest" "github.com/coredns/coredns/plugin/test" @@ -142,88 +141,3 @@ func difference(testRRs []dns.RR, gotRRs []dns.RR) []dns.RR { } return foundRRs } - -func TestEndpointsEquivalent(t *testing.T) { - epA := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.4", Hostname: "foo"}}, - }}, - } - epB := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.4", Hostname: "foo"}}, - }}, - } - epC := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.5", Hostname: "foo"}}, - }}, - } - epD := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.5", Hostname: "foo"}}, - }, - { - Addresses: []object.EndpointAddress{{IP: "1.2.2.2", Hostname: "foofoo"}}, - }}, - } - epE := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.5", Hostname: "foo"}, {IP: "1.1.1.1"}}, - }}, - } - epF := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.4", Hostname: "foofoo"}}, - }}, - } - epG := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.4", Hostname: "foo"}}, - Ports: []object.EndpointPort{{Name: "http", Port: 80, Protocol: "TCP"}}, - }}, - } - epH := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.4", Hostname: "foo"}}, - Ports: []object.EndpointPort{{Name: "newportname", Port: 80, Protocol: "TCP"}}, - }}, - } - epI := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.4", Hostname: "foo"}}, - Ports: []object.EndpointPort{{Name: "http", Port: 8080, Protocol: "TCP"}}, - }}, - } - epJ := object.Endpoints{ - Subsets: []object.EndpointSubset{{ - Addresses: []object.EndpointAddress{{IP: "1.2.3.4", Hostname: "foo"}}, - Ports: []object.EndpointPort{{Name: "http", Port: 80, Protocol: "UDP"}}, - }}, - } - - tests := []struct { - equiv bool - a *object.Endpoints - b *object.Endpoints - }{ - {true, &epA, &epB}, - {false, &epA, &epC}, - {false, &epA, &epD}, - {false, &epA, &epE}, - {false, &epA, &epF}, - {false, &epF, &epG}, - {false, &epG, &epH}, - {false, &epG, &epI}, - {false, &epG, &epJ}, - } - - for i, tc := range tests { - if tc.equiv && !endpointsEquivalent(tc.a, tc.b) { - t.Errorf("Test %d: expected endpoints to be equivalent and they are not.", i) - } - if !tc.equiv && endpointsEquivalent(tc.a, tc.b) { - t.Errorf("Test %d: expected endpoints to be seen as different but they were not.", i) - } - } -} diff --git a/plugin/pkg/watch/watch.go b/plugin/pkg/watch/watch.go deleted file mode 100644 index 7e77bb7b3..000000000 --- a/plugin/pkg/watch/watch.go +++ /dev/null @@ -1,23 +0,0 @@ -package watch - -// Chan is used to inform the server of a change. Whenever -// a watched FQDN has a change in data, that FQDN should be -// sent down this channel. -type Chan chan string - -// Watchable is the interface watchable plugins should implement -type Watchable interface { - // Name returns the plugin name. - Name() string - - // SetWatchChan is called when the watch channel is created. - SetWatchChan(Chan) - - // Watch is called whenever a watch is created for a FQDN. Plugins - // should send the FQDN down the watch channel when its data may have - // changed. This is an exact match only. - Watch(qname string) error - - // StopWatching is called whenever all watches are canceled for a FQDN. - StopWatching(qname string) -} diff --git a/plugin/pkg/watch/watcher.go b/plugin/pkg/watch/watcher.go deleted file mode 100644 index 86a952db2..000000000 --- a/plugin/pkg/watch/watcher.go +++ /dev/null @@ -1,178 +0,0 @@ -package watch - -import ( - "fmt" - "io" - "sync" - - "github.com/miekg/dns" - - "github.com/coredns/coredns/pb" - "github.com/coredns/coredns/plugin" - "github.com/coredns/coredns/plugin/pkg/log" - "github.com/coredns/coredns/request" -) - -// Watcher handles watch creation, cancellation, and processing. -type Watcher interface { - // Watch monitors a client stream and creates and cancels watches. - Watch(pb.DnsService_WatchServer) error - - // Stop cancels open watches and stops the watch processing go routine. - Stop() -} - -// Manager contains all the data needed to manage watches -type Manager struct { - changes Chan - stopper chan bool - counter int64 - watches map[string]watchlist - plugins []Watchable - mutex sync.Mutex -} - -type watchlist map[int64]pb.DnsService_WatchServer - -// NewWatcher creates a Watcher, which is used to manage watched names. -func NewWatcher(plugins []Watchable) *Manager { - w := &Manager{changes: make(Chan), stopper: make(chan bool), watches: make(map[string]watchlist), plugins: plugins} - - for _, p := range plugins { - p.SetWatchChan(w.changes) - } - - go w.process() - return w -} - -func (w *Manager) nextID() int64 { - w.mutex.Lock() - - w.counter++ - id := w.counter - - w.mutex.Unlock() - return id -} - -// Watch monitors a client stream and creates and cancels watches. -func (w *Manager) Watch(stream pb.DnsService_WatchServer) error { - for { - in, err := stream.Recv() - if err == io.EOF { - return nil - } - if err != nil { - return err - } - create := in.GetCreateRequest() - if create != nil { - msg := new(dns.Msg) - err := msg.Unpack(create.Query.Msg) - if err != nil { - log.Warningf("Could not decode watch request: %s\n", err) - stream.Send(&pb.WatchResponse{Err: "could not decode request"}) - continue - } - id := w.nextID() - if err := stream.Send(&pb.WatchResponse{WatchId: id, Created: true}); err != nil { - // if we fail to notify client of watch creation, don't create the watch - continue - } - - // Normalize qname - qname := (&request.Request{Req: msg}).Name() - - w.mutex.Lock() - if _, ok := w.watches[qname]; !ok { - w.watches[qname] = make(watchlist) - } - w.watches[qname][id] = stream - w.mutex.Unlock() - - for _, p := range w.plugins { - err := p.Watch(qname) - if err != nil { - log.Warningf("Failed to start watch for %s in plugin %s: %s\n", qname, p.Name(), err) - stream.Send(&pb.WatchResponse{Err: fmt.Sprintf("failed to start watch for %s in plugin %s", qname, p.Name())}) - } - } - continue - } - - cancel := in.GetCancelRequest() - if cancel != nil { - w.mutex.Lock() - for qname, wl := range w.watches { - ws, ok := wl[cancel.WatchId] - if !ok { - continue - } - - // only allow cancels from the client that started it - // TODO: test what happens if a stream tries to cancel a watchID that it doesn't own - if ws != stream { - continue - } - - delete(wl, cancel.WatchId) - - // if there are no more watches for this qname, we should tell the plugins - if len(wl) == 0 { - for _, p := range w.plugins { - p.StopWatching(qname) - } - delete(w.watches, qname) - } - - // let the client know we canceled the watch - stream.Send(&pb.WatchResponse{WatchId: cancel.WatchId, Canceled: true}) - } - w.mutex.Unlock() - continue - } - } -} - -func (w *Manager) process() { - for { - select { - case <-w.stopper: - return - case changed := <-w.changes: - w.mutex.Lock() - for qname, wl := range w.watches { - if plugin.Zones([]string{changed}).Matches(qname) == "" { - continue - } - for id, stream := range wl { - wr := pb.WatchResponse{WatchId: id, Qname: qname} - err := stream.Send(&wr) - if err != nil { - log.Warningf("Error sending change for %s to watch %d: %s. Removing watch.\n", qname, id, err) - delete(w.watches[qname], id) - } - } - } - w.mutex.Unlock() - } - } -} - -// Stop cancels open watches and stops the watch processing go routine. -func (w *Manager) Stop() { - w.stopper <- true - w.mutex.Lock() - for wn, wl := range w.watches { - for id, stream := range wl { - wr := pb.WatchResponse{WatchId: id, Canceled: true} - err := stream.Send(&wr) - if err != nil { - log.Warningf("Error notifying client of cancellation: %s\n", err) - } - } - delete(w.watches, wn) - } - w.mutex.Unlock() -}