micro/runtime/service/service.go

286 lines
5.6 KiB
Go
Raw Normal View History

2019-11-20 17:54:42 +03:00
package service
import (
"context"
"sync"
"github.com/micro/go-micro/v2/client"
"github.com/micro/go-micro/v2/runtime"
pb "github.com/micro/go-micro/v2/runtime/service/proto"
2019-11-20 17:54:42 +03:00
)
type svc struct {
sync.RWMutex
options runtime.Options
runtime pb.RuntimeService
}
// Init initializes runtime with given options
func (s *svc) Init(opts ...runtime.Option) error {
s.Lock()
defer s.Unlock()
for _, o := range opts {
o(&s.options)
}
return nil
}
// Create registers a service in the runtime
func (s *svc) Create(svc *runtime.Service, opts ...runtime.CreateOption) error {
var options runtime.CreateOptions
2019-11-20 17:54:42 +03:00
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.Background()
}
2019-11-20 17:54:42 +03:00
// set the default source from MICRO_RUNTIME_SOURCE
if len(svc.Source) == 0 {
svc.Source = s.options.Source
}
2019-11-20 17:54:42 +03:00
// runtime service create request
req := &pb.CreateRequest{
Service: &pb.Service{
Name: svc.Name,
Version: svc.Version,
Source: svc.Source,
Metadata: svc.Metadata,
2019-11-20 17:54:42 +03:00
},
Options: &pb.CreateOptions{
Command: options.Command,
Args: options.Args,
Env: options.Env,
Type: options.Type,
Image: options.Image,
2019-11-20 17:54:42 +03:00
},
}
if _, err := s.runtime.Create(options.Context, req); err != nil {
2019-11-20 17:54:42 +03:00
return err
}
return nil
}
func (s *svc) Logs(service *runtime.Service, opts ...runtime.LogsOption) (runtime.LogStream, error) {
var options runtime.LogsOptions
for _, o := range opts {
o(&options)
}
2020-04-28 10:49:39 +03:00
if options.Context == nil {
options.Context = context.Background()
}
ls, err := s.runtime.Logs(options.Context, &pb.LogsRequest{
Service: service.Name,
2020-04-28 10:49:39 +03:00
Stream: options.Stream,
Count: options.Count,
})
if err != nil {
return nil, err
}
logStream := &serviceLogStream{
service: service.Name,
stream: make(chan runtime.LogRecord),
stop: make(chan bool),
}
2020-04-28 10:49:39 +03:00
go func() {
for {
select {
// @todo this never seems to return, investigate
case <-ls.Context().Done():
logStream.Stop()
}
}
}()
go func() {
for {
2020-04-28 10:49:39 +03:00
select {
// @todo this never seems to return, investigate
case <-ls.Context().Done():
return
case _, ok := <-logStream.stream:
if !ok {
return
}
default:
record := pb.LogRecord{}
err := ls.RecvMsg(&record)
if err != nil {
logStream.Stop()
return
}
logStream.stream <- runtime.LogRecord{
Message: record.GetMessage(),
Metadata: record.GetMetadata(),
}
}
}
}()
return logStream, nil
}
type serviceLogStream struct {
service string
stream chan runtime.LogRecord
sync.Mutex
stop chan bool
err error
}
func (l *serviceLogStream) Error() error {
return l.err
}
func (l *serviceLogStream) Chan() chan runtime.LogRecord {
return l.stream
}
func (l *serviceLogStream) Stop() error {
l.Lock()
defer l.Unlock()
select {
case <-l.stop:
return nil
default:
2020-04-28 10:49:39 +03:00
close(l.stream)
close(l.stop)
}
return nil
}
// Read returns the service with the given name from the runtime
2019-11-29 14:35:00 +03:00
func (s *svc) Read(opts ...runtime.ReadOption) ([]*runtime.Service, error) {
var options runtime.ReadOptions
2019-11-20 17:54:42 +03:00
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.Background()
}
2019-11-20 17:54:42 +03:00
// runtime service create request
req := &pb.ReadRequest{
Options: &pb.ReadOptions{
2019-11-29 14:35:00 +03:00
Service: options.Service,
2019-11-20 17:54:42 +03:00
Version: options.Version,
2019-11-29 14:35:00 +03:00
Type: options.Type,
2019-11-20 17:54:42 +03:00
},
}
resp, err := s.runtime.Read(options.Context, req)
2019-11-20 17:54:42 +03:00
if err != nil {
return nil, err
}
services := make([]*runtime.Service, 0, len(resp.Services))
for _, service := range resp.Services {
svc := &runtime.Service{
Name: service.Name,
Version: service.Version,
Source: service.Source,
Metadata: service.Metadata,
2019-11-20 17:54:42 +03:00
}
services = append(services, svc)
}
return services, nil
}
// Update updates the running service
func (s *svc) Update(svc *runtime.Service, opts ...runtime.UpdateOption) error {
var options runtime.UpdateOptions
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.Background()
}
2019-11-20 17:54:42 +03:00
// runtime service create request
req := &pb.UpdateRequest{
Service: &pb.Service{
Name: svc.Name,
Version: svc.Version,
Source: svc.Source,
Metadata: svc.Metadata,
2019-11-20 17:54:42 +03:00
},
}
if _, err := s.runtime.Update(options.Context, req); err != nil {
2019-11-20 17:54:42 +03:00
return err
}
return nil
}
// Delete stops and removes the service from the runtime
func (s *svc) Delete(svc *runtime.Service, opts ...runtime.DeleteOption) error {
var options runtime.DeleteOptions
for _, o := range opts {
o(&options)
}
if options.Context == nil {
options.Context = context.Background()
}
2019-11-20 17:54:42 +03:00
// runtime service create request
req := &pb.DeleteRequest{
Service: &pb.Service{
Name: svc.Name,
Version: svc.Version,
Source: svc.Source,
Metadata: svc.Metadata,
2019-11-20 17:54:42 +03:00
},
}
if _, err := s.runtime.Delete(options.Context, req); err != nil {
2019-11-20 17:54:42 +03:00
return err
}
return nil
}
// Start starts the runtime
func (s *svc) Start() error {
// NOTE: nothing to be done here
return nil
}
// Stop stops the runtime
func (s *svc) Stop() error {
// NOTE: nothing to be done here
return nil
}
// Returns the runtime service implementation
func (s *svc) String() string {
return "service"
}
2020-01-19 03:55:01 +03:00
// NewRuntime creates new service runtime and returns it
func NewRuntime(opts ...runtime.Option) runtime.Runtime {
// get default options
options := runtime.Options{}
// apply requested options
for _, o := range opts {
o(&options)
}
// create default client
cli := client.DefaultClient
return &svc{
options: options,
runtime: pb.NewRuntimeService(runtime.DefaultName, cli),
}
}