diff --git a/README.md b/README.md index 3b038e8..3cdecea 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,10 @@ Please, take into account the educational status of this project and DO NOT use `-server` -- Run Orkestrator in the server mode. +### Worker options + +`-worker` -- Run Orkestrator in the worker mode. + ## Specification Orkestrator consists of a few components that interact with each other to manage containers: diff --git a/api/proto/worker/worker.pb.go b/api/proto/worker/worker.pb.go new file mode 100644 index 0000000..ee38250 --- /dev/null +++ b/api/proto/worker/worker.pb.go @@ -0,0 +1,445 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.31.0 +// protoc v4.25.1 +// source: worker.proto + +package worker + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type RegisterWorkerRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` +} + +func (x *RegisterWorkerRequest) Reset() { + *x = RegisterWorkerRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_worker_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *RegisterWorkerRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RegisterWorkerRequest) ProtoMessage() {} + +func (x *RegisterWorkerRequest) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[0] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RegisterWorkerRequest.ProtoReflect.Descriptor instead. +func (*RegisterWorkerRequest) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{0} +} + +func (x *RegisterWorkerRequest) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +type RegisterWorkerResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` +} + +func (x *RegisterWorkerResponse) Reset() { + *x = RegisterWorkerResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_worker_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *RegisterWorkerResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RegisterWorkerResponse) ProtoMessage() {} + +func (x *RegisterWorkerResponse) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RegisterWorkerResponse.ProtoReflect.Descriptor instead. +func (*RegisterWorkerResponse) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{1} +} + +func (x *RegisterWorkerResponse) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *RegisterWorkerResponse) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +type UnsubscribeRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields +} + +func (x *UnsubscribeRequest) Reset() { + *x = UnsubscribeRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_worker_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *UnsubscribeRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*UnsubscribeRequest) ProtoMessage() {} + +func (x *UnsubscribeRequest) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[2] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use UnsubscribeRequest.ProtoReflect.Descriptor instead. +func (*UnsubscribeRequest) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{2} +} + +type UnsubscribeResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields +} + +func (x *UnsubscribeResponse) Reset() { + *x = UnsubscribeResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_worker_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *UnsubscribeResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*UnsubscribeResponse) ProtoMessage() {} + +func (x *UnsubscribeResponse) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[3] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use UnsubscribeResponse.ProtoReflect.Descriptor instead. +func (*UnsubscribeResponse) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{3} +} + +type StatusRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields +} + +func (x *StatusRequest) Reset() { + *x = StatusRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_worker_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *StatusRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StatusRequest) ProtoMessage() {} + +func (x *StatusRequest) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[4] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StatusRequest.ProtoReflect.Descriptor instead. +func (*StatusRequest) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{4} +} + +type StatusResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields +} + +func (x *StatusResponse) Reset() { + *x = StatusResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_worker_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *StatusResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StatusResponse) ProtoMessage() {} + +func (x *StatusResponse) ProtoReflect() protoreflect.Message { + mi := &file_worker_proto_msgTypes[5] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StatusResponse.ProtoReflect.Descriptor instead. +func (*StatusResponse) Descriptor() ([]byte, []int) { + return file_worker_proto_rawDescGZIP(), []int{5} +} + +var File_worker_proto protoreflect.FileDescriptor + +var file_worker_proto_rawDesc = []byte{ + 0x0a, 0x0c, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x06, + 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x22, 0x2b, 0x0a, 0x15, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, + 0x65, 0x72, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, + 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, + 0x61, 0x6d, 0x65, 0x22, 0x3c, 0x0a, 0x16, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x57, + 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x0e, 0x0a, + 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x12, 0x0a, + 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, + 0x65, 0x22, 0x14, 0x0a, 0x12, 0x55, 0x6e, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, + 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x15, 0x0a, 0x13, 0x55, 0x6e, 0x73, 0x75, 0x62, + 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x0f, + 0x0a, 0x0d, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, + 0x10, 0x0a, 0x0e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, + 0x65, 0x32, 0xdf, 0x01, 0x0a, 0x06, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12, 0x4f, 0x0a, 0x0e, + 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x12, 0x1d, + 0x2e, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, + 0x57, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1e, 0x2e, + 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x52, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x57, + 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x4b, 0x0a, + 0x10, 0x55, 0x6e, 0x72, 0x65, 0x67, 0x69, 0x73, 0x74, 0x65, 0x72, 0x57, 0x6f, 0x72, 0x6b, 0x65, + 0x72, 0x12, 0x1a, 0x2e, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x55, 0x6e, 0x73, 0x75, 0x62, + 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1b, 0x2e, + 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x55, 0x6e, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, + 0x62, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x37, 0x0a, 0x06, 0x53, 0x74, + 0x61, 0x74, 0x75, 0x73, 0x12, 0x15, 0x2e, 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x53, 0x74, + 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x77, 0x6f, + 0x72, 0x6b, 0x65, 0x72, 0x2e, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, + 0x6e, 0x73, 0x65, 0x42, 0x31, 0x5a, 0x2f, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, + 0x6d, 0x2f, 0x61, 0x72, 0x79, 0x62, 0x6f, 0x6c, 0x6f, 0x76, 0x6c, 0x65, 0x76, 0x2f, 0x6f, 0x72, + 0x6b, 0x65, 0x73, 0x74, 0x72, 0x61, 0x74, 0x6f, 0x72, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, + 0x77, 0x6f, 0x72, 0x6b, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, +} + +var ( + file_worker_proto_rawDescOnce sync.Once + file_worker_proto_rawDescData = file_worker_proto_rawDesc +) + +func file_worker_proto_rawDescGZIP() []byte { + file_worker_proto_rawDescOnce.Do(func() { + file_worker_proto_rawDescData = protoimpl.X.CompressGZIP(file_worker_proto_rawDescData) + }) + return file_worker_proto_rawDescData +} + +var file_worker_proto_msgTypes = make([]protoimpl.MessageInfo, 6) +var file_worker_proto_goTypes = []interface{}{ + (*RegisterWorkerRequest)(nil), // 0: worker.RegisterWorkerRequest + (*RegisterWorkerResponse)(nil), // 1: worker.RegisterWorkerResponse + (*UnsubscribeRequest)(nil), // 2: worker.UnsubscribeRequest + (*UnsubscribeResponse)(nil), // 3: worker.UnsubscribeResponse + (*StatusRequest)(nil), // 4: worker.StatusRequest + (*StatusResponse)(nil), // 5: worker.StatusResponse +} +var file_worker_proto_depIdxs = []int32{ + 0, // 0: worker.Worker.RegisterWorker:input_type -> worker.RegisterWorkerRequest + 2, // 1: worker.Worker.UnregisterWorker:input_type -> worker.UnsubscribeRequest + 4, // 2: worker.Worker.Status:input_type -> worker.StatusRequest + 1, // 3: worker.Worker.RegisterWorker:output_type -> worker.RegisterWorkerResponse + 3, // 4: worker.Worker.UnregisterWorker:output_type -> worker.UnsubscribeResponse + 5, // 5: worker.Worker.Status:output_type -> worker.StatusResponse + 3, // [3:6] is the sub-list for method output_type + 0, // [0:3] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_worker_proto_init() } +func file_worker_proto_init() { + if File_worker_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_worker_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*RegisterWorkerRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_worker_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*RegisterWorkerResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_worker_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*UnsubscribeRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_worker_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*UnsubscribeResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_worker_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*StatusRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_worker_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*StatusResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_worker_proto_rawDesc, + NumEnums: 0, + NumMessages: 6, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_worker_proto_goTypes, + DependencyIndexes: file_worker_proto_depIdxs, + MessageInfos: file_worker_proto_msgTypes, + }.Build() + File_worker_proto = out.File + file_worker_proto_rawDesc = nil + file_worker_proto_goTypes = nil + file_worker_proto_depIdxs = nil +} diff --git a/api/proto/worker/worker.proto b/api/proto/worker/worker.proto index 17eb435..b68eac0 100644 --- a/api/proto/worker/worker.proto +++ b/api/proto/worker/worker.proto @@ -4,8 +4,13 @@ package worker; option go_package = "github.com/arybolovlev/orkestrator/proto/worker"; -message SubscribeRequest {} -message SubscribeResponse {} +message RegisterWorkerRequest { + string name = 1; +} +message RegisterWorkerResponse { + string id = 1; + string name = 2; +} message UnsubscribeRequest {} message UnsubscribeResponse {} @@ -14,7 +19,7 @@ message StatusRequest {} message StatusResponse {} service Worker { - rpc Subscribe(SubscribeRequest) returns (stream SubscribeResponse); - rpc Unsubscribe(UnsubscribeRequest) returns (UnsubscribeResponse); + rpc RegisterWorker(RegisterWorkerRequest) returns (RegisterWorkerResponse); + rpc UnregisterWorker(UnsubscribeRequest) returns (UnsubscribeResponse); rpc Status(StatusRequest) returns (StatusResponse); } diff --git a/api/proto/worker/worker_grpc.pb.go b/api/proto/worker/worker_grpc.pb.go new file mode 100644 index 0000000..1618fde --- /dev/null +++ b/api/proto/worker/worker_grpc.pb.go @@ -0,0 +1,183 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.3.0 +// - protoc v4.25.1 +// source: worker.proto + +package worker + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.32.0 or later. +const _ = grpc.SupportPackageIsVersion7 + +const ( + Worker_RegisterWorker_FullMethodName = "/worker.Worker/RegisterWorker" + Worker_UnregisterWorker_FullMethodName = "/worker.Worker/UnregisterWorker" + Worker_Status_FullMethodName = "/worker.Worker/Status" +) + +// WorkerClient is the client API for Worker service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type WorkerClient interface { + RegisterWorker(ctx context.Context, in *RegisterWorkerRequest, opts ...grpc.CallOption) (*RegisterWorkerResponse, error) + UnregisterWorker(ctx context.Context, in *UnsubscribeRequest, opts ...grpc.CallOption) (*UnsubscribeResponse, error) + Status(ctx context.Context, in *StatusRequest, opts ...grpc.CallOption) (*StatusResponse, error) +} + +type workerClient struct { + cc grpc.ClientConnInterface +} + +func NewWorkerClient(cc grpc.ClientConnInterface) WorkerClient { + return &workerClient{cc} +} + +func (c *workerClient) RegisterWorker(ctx context.Context, in *RegisterWorkerRequest, opts ...grpc.CallOption) (*RegisterWorkerResponse, error) { + out := new(RegisterWorkerResponse) + err := c.cc.Invoke(ctx, Worker_RegisterWorker_FullMethodName, in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *workerClient) UnregisterWorker(ctx context.Context, in *UnsubscribeRequest, opts ...grpc.CallOption) (*UnsubscribeResponse, error) { + out := new(UnsubscribeResponse) + err := c.cc.Invoke(ctx, Worker_UnregisterWorker_FullMethodName, in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *workerClient) Status(ctx context.Context, in *StatusRequest, opts ...grpc.CallOption) (*StatusResponse, error) { + out := new(StatusResponse) + err := c.cc.Invoke(ctx, Worker_Status_FullMethodName, in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + +// WorkerServer is the server API for Worker service. +// All implementations must embed UnimplementedWorkerServer +// for forward compatibility +type WorkerServer interface { + RegisterWorker(context.Context, *RegisterWorkerRequest) (*RegisterWorkerResponse, error) + UnregisterWorker(context.Context, *UnsubscribeRequest) (*UnsubscribeResponse, error) + Status(context.Context, *StatusRequest) (*StatusResponse, error) + mustEmbedUnimplementedWorkerServer() +} + +// UnimplementedWorkerServer must be embedded to have forward compatible implementations. +type UnimplementedWorkerServer struct { +} + +func (UnimplementedWorkerServer) RegisterWorker(context.Context, *RegisterWorkerRequest) (*RegisterWorkerResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method RegisterWorker not implemented") +} +func (UnimplementedWorkerServer) UnregisterWorker(context.Context, *UnsubscribeRequest) (*UnsubscribeResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method UnregisterWorker not implemented") +} +func (UnimplementedWorkerServer) Status(context.Context, *StatusRequest) (*StatusResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method Status not implemented") +} +func (UnimplementedWorkerServer) mustEmbedUnimplementedWorkerServer() {} + +// UnsafeWorkerServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to WorkerServer will +// result in compilation errors. +type UnsafeWorkerServer interface { + mustEmbedUnimplementedWorkerServer() +} + +func RegisterWorkerServer(s grpc.ServiceRegistrar, srv WorkerServer) { + s.RegisterService(&Worker_ServiceDesc, srv) +} + +func _Worker_RegisterWorker_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(RegisterWorkerRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(WorkerServer).RegisterWorker(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Worker_RegisterWorker_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(WorkerServer).RegisterWorker(ctx, req.(*RegisterWorkerRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Worker_UnregisterWorker_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(UnsubscribeRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(WorkerServer).UnregisterWorker(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Worker_UnregisterWorker_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(WorkerServer).UnregisterWorker(ctx, req.(*UnsubscribeRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Worker_Status_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(StatusRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(WorkerServer).Status(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Worker_Status_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(WorkerServer).Status(ctx, req.(*StatusRequest)) + } + return interceptor(ctx, in, info, handler) +} + +// Worker_ServiceDesc is the grpc.ServiceDesc for Worker service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var Worker_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "worker.Worker", + HandlerType: (*WorkerServer)(nil), + Methods: []grpc.MethodDesc{ + { + MethodName: "RegisterWorker", + Handler: _Worker_RegisterWorker_Handler, + }, + { + MethodName: "UnregisterWorker", + Handler: _Worker_UnregisterWorker_Handler, + }, + { + MethodName: "Status", + Handler: _Worker_Status_Handler, + }, + }, + Streams: []grpc.StreamDesc{}, + Metadata: "worker.proto", +} diff --git a/api/job/job.go b/api/structs/job.go similarity index 97% rename from api/job/job.go rename to api/structs/job.go index 3c7b6e5..44dfa41 100644 --- a/api/job/job.go +++ b/api/structs/job.go @@ -1,4 +1,4 @@ -package job +package structs import ( "errors" diff --git a/api/job/job_test.go b/api/structs/job_test.go similarity index 98% rename from api/job/job_test.go rename to api/structs/job_test.go index 83349cb..f4247a6 100644 --- a/api/job/job_test.go +++ b/api/structs/job_test.go @@ -1,4 +1,4 @@ -package job +package structs import ( "testing" diff --git a/api/structs/worker.go b/api/structs/worker.go new file mode 100644 index 0000000..b66f0e7 --- /dev/null +++ b/api/structs/worker.go @@ -0,0 +1,6 @@ +package structs + +type Worker struct { + ID string + Name string +} diff --git a/client/client.go b/client/client.go index b308a3d..a6d2f1f 100644 --- a/client/client.go +++ b/client/client.go @@ -9,12 +9,12 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" - "github.com/arybolovlev/orkestrator/api/job" "github.com/arybolovlev/orkestrator/api/proto/client" + "github.com/arybolovlev/orkestrator/api/structs" ) type JobSpec struct { - Job []job.Job `hcl:"job,block"` + Job []structs.Job `hcl:"job,block"` } func Run(port int, spec string) { @@ -54,6 +54,6 @@ func Run(port int, spec string) { log.Fatalf("falied to create job %s: %s", req.Name, err) os.Exit(1) } - log.Println("New job sucessfully created:", job) + log.Println("new job sucessfully created:", job) } } diff --git a/cmd/main.go b/cmd/main.go index 25fb603..2b5f742 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -6,6 +6,7 @@ import ( "github.com/arybolovlev/orkestrator/client" "github.com/arybolovlev/orkestrator/manager" + "github.com/arybolovlev/orkestrator/worker" ) func main() { @@ -17,6 +18,12 @@ func main() { var srv bool flag.BoolVar(&srv, "server", false, "Run Orkestrator in the server mode") + // Worker options + var wrk bool + flag.BoolVar(&wrk, "worker", false, "Run Orkestrator in the worker mode") + var wrkName string + flag.StringVar(&wrkName, "worker-name", "", "Worker name") + // Client options var cli bool flag.BoolVar(&cli, "client", false, "Run Orkestrator in the client mode") @@ -30,6 +37,11 @@ func main() { manager.Run(port) } + if wrk { + log.Println("Running Orkestrator Worker") + worker.Run(wrkName, port) + } + if cli { log.Println("Running Orkestrator Client") client.Run(port, file) diff --git a/manager/manager_client.go b/manager/client.go similarity index 79% rename from manager/manager_client.go rename to manager/client.go index 0013a8f..70215cd 100644 --- a/manager/manager_client.go +++ b/manager/client.go @@ -6,19 +6,19 @@ import ( "github.com/google/uuid" - "github.com/arybolovlev/orkestrator/api/job" "github.com/arybolovlev/orkestrator/api/proto/client" + "github.com/arybolovlev/orkestrator/api/structs" "github.com/arybolovlev/orkestrator/api/task" ) -type manager struct { +type clientManager struct { client.ClientServer } -func (m *manager) RegisterJob(ctx context.Context, req *client.RegisterJobRequest) (*client.RegisterJobResponse, error) { +func (m *clientManager) RegisterJob(ctx context.Context, req *client.RegisterJobRequest) (*client.RegisterJobResponse, error) { log.Println("new job request received:", req.Name) - j := job.Job{Name: req.Name} + j := structs.Job{Name: req.Name} if _, ok := Jobs[req.Name]; ok { log.Printf("job %s already exists\n", j.Name) j.ID = Jobs[req.Name].ID @@ -43,7 +43,7 @@ func (m *manager) RegisterJob(ctx context.Context, req *client.RegisterJobReques } log.Printf("job %s is valid", j.Name) - Jobs[j.Name] = j + Jobs[req.Name] = j log.Println("new job was successfully registered:", j.ID, j.Name) log.Println("total jobs registered:", len(Jobs)) diff --git a/manager/manager.go b/manager/manager.go index a668d74..e952080 100644 --- a/manager/manager.go +++ b/manager/manager.go @@ -7,12 +7,14 @@ import ( "google.golang.org/grpc" - "github.com/arybolovlev/orkestrator/api/job" "github.com/arybolovlev/orkestrator/api/proto/client" + "github.com/arybolovlev/orkestrator/api/proto/worker" + "github.com/arybolovlev/orkestrator/api/structs" ) var ( - Jobs = map[string]job.Job{} + Jobs = map[string]structs.Job{} + Workers = map[string]structs.Worker{} ) func Run(port int) { @@ -22,7 +24,8 @@ func Run(port int) { } gs := grpc.NewServer() - client.RegisterClientServer(gs, &manager{}) + client.RegisterClientServer(gs, &clientManager{}) + worker.RegisterWorkerServer(gs, &workerManager{}) log.Printf("server listening at %v", ln.Addr()) diff --git a/manager/worker.go b/manager/worker.go new file mode 100644 index 0000000..583ab69 --- /dev/null +++ b/manager/worker.go @@ -0,0 +1,37 @@ +package manager + +import ( + "context" + "log" + + "github.com/google/uuid" + + "github.com/arybolovlev/orkestrator/api/proto/worker" + "github.com/arybolovlev/orkestrator/api/structs" +) + +type workerManager struct { + worker.WorkerServer +} + +func (m *workerManager) RegisterWorker(ctx context.Context, req *worker.RegisterWorkerRequest) (*worker.RegisterWorkerResponse, error) { + log.Println("new worker registry received:", req.Name) + + w := structs.Worker{Name: req.Name} + if _, ok := Workers[req.Name]; ok { + log.Printf("worker %s already registered\n", w.Name) + w.ID = Workers[req.Name].ID + } else { + w.ID = uuid.New().String() + log.Printf("register a new worker %s: %s\n", w.Name, w.ID) + log.Println("new worker was successfully registered:", w.ID, w.Name) + } + + Workers[req.Name] = w + log.Println("total workers registered:", len(Workers)) + + return &worker.RegisterWorkerResponse{ + Id: w.ID, + Name: w.Name, + }, nil +} diff --git a/worker/worker.go b/worker/worker.go new file mode 100644 index 0000000..69c5a42 --- /dev/null +++ b/worker/worker.go @@ -0,0 +1,29 @@ +package worker + +import ( + "context" + "fmt" + "log" + "os" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/arybolovlev/orkestrator/api/proto/worker" +) + +func Run(name string, port int) { + conn, err := grpc.Dial(fmt.Sprintf("localhost:%d", port), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + log.Fatalf("did not connect: %v", err) + } + defer conn.Close() + w := worker.NewWorkerClient(conn) + + wrk, err := w.RegisterWorker(context.Background(), &worker.RegisterWorkerRequest{Name: name}) + if err != nil { + log.Fatalf("falied to subscribe worker %s: %s", name, err) + os.Exit(1) + } + log.Println("new worker sucessfully subscribed:", wrk) +}