go/go-micro
1
0
Fork 0
mirror of https://github.com/go-micro/go-micro.git synced 2025-06-24 22:26:54 +02:00
Code Issues Releases Activity

Merge branch 'master' into git-secrets

Browse Source
This commit is contained in:
ben-toogood
2020-04-23 15:01:47 +01:00
committed by GitHub
parent b5f53595ca 692b27578c
commit b692c045b5
14 changed files with 411 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
runtime/kubernetes/kubernetes.go
View File

@ -6,7 +6,7 @@ import (
"sync"
"time"
"github.com/micro/go-micro/v2/logger"
log "github.com/micro/go-micro/v2/logger"
"github.com/micro/go-micro/v2/runtime"
"github.com/micro/go-micro/v2/util/kubernetes/client"
)
@ -24,11 +24,48 @@ type kubernetes struct {
closed chan bool
// client is kubernetes client
client client.Client
// namespaces which exist
namespaces []client.Namespace
}
// namespaceExists returns a boolean indicating if a namespace exists
func (k *kubernetes) namespaceExists(name string) (bool, error) {
// populate the cache
if k.namespaces == nil {
namespaceList := new(client.NamespaceList)
resource := &client.Resource{Kind: "namespace", Value: namespaceList}
if err := k.client.List(resource); err != nil {
return false, err
}
k.namespaces = namespaceList.Items
}
// check if the namespace exists in the cache
for _, n := range k.namespaces {
if n.Metadata.Name == name {
return true, nil
}
}
return false, nil
}
// createNamespace creates a new k8s namespace
func (k *kubernetes) createNamespace(namespace string) error {
ns := client.Namespace{Metadata: &client.Metadata{Name: namespace}}
err := k.client.Create(&client.Resource{Kind: "namespace", Value: ns})
// add to cache
if err == nil && k.namespaces != nil {
k.namespaces = append(k.namespaces, ns)
}
return err
}
// getService queries kubernetes for micro service
// NOTE: this function is not thread-safe
func (k *kubernetes) getService(labels map[string]string) ([]*service, error) {
func (k *kubernetes) getService(labels map[string]string, opts ...client.GetOption) ([]*service, error) {
// get the service status
serviceList := new(client.ServiceList)
r := &client.Resource{
@ -36,8 +73,10 @@ func (k *kubernetes) getService(labels map[string]string) ([]*service, error) {
Value: serviceList,
}
opts = append(opts, client.GetLabels(labels))
// get the service from k8s
if err := k.client.Get(r, labels); err != nil {
if err := k.client.Get(r, opts...); err != nil {
return nil, err
}
@ -47,7 +86,7 @@ func (k *kubernetes) getService(labels map[string]string) ([]*service, error) {
Kind: "deployment",
Value: depList,
}
if err := k.client.Get(d, labels); err != nil {
if err := k.client.Get(d, opts...); err != nil {
return nil, err
}
@ -57,7 +96,7 @@ func (k *kubernetes) getService(labels map[string]string) ([]*service, error) {
Kind: "pod",
Value: podList,
}
if err := k.client.Get(p, labels); err != nil {
if err := k.client.Get(p, opts...); err != nil {
return nil, err
}
@ -206,8 +245,8 @@ func (k *kubernetes) run(events <-chan runtime.Event) {
// - do we even need the ticker for k8s services?
case event := <-events:
// NOTE: we only handle Update events for now
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime received notification event: %v", event)
if log.V(log.DebugLevel, log.DefaultLogger) {
log.Debugf("Runtime received notification event: %v", event)
}
switch event.Type {
case runtime.Update:
@ -237,11 +276,11 @@ func (k *kubernetes) run(events <-chan runtime.Event) {
err := k.client.Get(&client.Resource{
Kind: "deployment",
Value: deployed,
}, labels)
}, client.GetLabels(labels))
if err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime update failed to get service %s: %v", event.Service, err)
if log.V(log.DebugLevel, log.DefaultLogger) {
log.Debugf("Runtime update failed to get service %s: %v", event.Service, err)
}
continue
}
@ -261,20 +300,20 @@ func (k *kubernetes) run(events <-chan runtime.Event) {
// update the build time
service.Spec.Template.Metadata.Annotations["updated"] = fmt.Sprintf("%d", event.Timestamp.Unix())
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime updating service: %s deployment: %s", event.Service, service.Metadata.Name)
if log.V(log.DebugLevel, log.DefaultLogger) {
log.Debugf("Runtime updating service: %s deployment: %s", event.Service, service.Metadata.Name)
}
if err := k.client.Update(deploymentResource(&service)); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime failed to update service %s: %v", event.Service, err)
if log.V(log.DebugLevel, log.DefaultLogger) {
log.Debugf("Runtime failed to update service %s: %v", event.Service, err)
}
continue
}
}
}
case <-k.closed:
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime stopped")
if log.V(log.DebugLevel, log.DefaultLogger) {
log.Debugf("Runtime stopped")
}
return
}
@ -304,7 +343,7 @@ func (k *kubernetes) Logs(s *runtime.Service, options ...runtime.LogsOption) (ru
go func() {
records, err := klo.Read()
if err != nil {
logger.Errorf("Failed to get logs for service '%v' from k8s: %v", err)
log.Errorf("Failed to get logs for service '%v' from k8s: %v", err)
return
}
// @todo: this might actually not run before podLogStream starts
@ -370,6 +409,16 @@ func (k *kubernetes) Create(s *runtime.Service, opts ...runtime.CreateOption) er
s.Source = k.options.Source
}
// ensure the namespace exists
namespace := client.SerializeResourceName(options.Namespace)
if exist, err := k.namespaceExists(namespace); err == nil && !exist {
if err := k.createNamespace(namespace); err != nil {
return err
}
} else if err != nil {
return err
}
// determine the image from the source and options
options.Image = k.getImage(s, options)
@ -377,7 +426,7 @@ func (k *kubernetes) Create(s *runtime.Service, opts ...runtime.CreateOption) er
service := newService(s, options)
// start the service
return service.Start(k.client)
return service.Start(k.client, client.CreateNamespace(options.Namespace))
}
// Read returns all instances of given service
@ -422,8 +471,12 @@ func (k *kubernetes) Read(opts ...runtime.ReadOption) ([]*runtime.Service, error
}
// Update the service in place
func (k *kubernetes) Update(s *runtime.Service) error {
// TODO: set the type
func (k *kubernetes) Update(s *runtime.Service, opts ...runtime.UpdateOption) error {
var options runtime.UpdateOptions
for _, o := range opts {
o(&options)
}
labels := map[string]string{}
if len(s.Name) > 0 {
@ -458,7 +511,7 @@ func (k *kubernetes) Update(s *runtime.Service) error {
service.kdeploy.Spec.Template.Metadata.Annotations["updated"] = fmt.Sprintf("%d", time.Now().Unix())
// update the service
if err := service.Update(k.client); err != nil {
if err := service.Update(k.client, client.UpdateNamespace(options.Namespace)); err != nil {
return err
}
}
@ -467,16 +520,22 @@ func (k *kubernetes) Update(s *runtime.Service) error {
}
// Delete removes a service
func (k *kubernetes) Delete(s *runtime.Service) error {
func (k *kubernetes) Delete(s *runtime.Service, opts ...runtime.DeleteOption) error {
var options runtime.DeleteOptions
for _, o := range opts {
o(&options)
}
k.Lock()
defer k.Unlock()
// create new kubernetes micro service
service := newService(s, runtime.CreateOptions{
Type: k.options.Type,
Type: k.options.Type,
Namespace: options.Namespace,
})
return service.Stop(k.client)
return service.Stop(k.client, client.DeleteNamespace(options.Namespace))
}
// Start starts the runtime
@ -499,8 +558,8 @@ func (k *kubernetes) Start() error {
events, err = k.options.Scheduler.Notify()
if err != nil {
// TODO: should we bail here?
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime failed to start update notifier")
if log.V(log.DebugLevel, log.DefaultLogger) {
log.Debugf("Runtime failed to start update notifier")
}
}
}

2
runtime/kubernetes/kubernetes_logs.go
View File

@ -70,7 +70,7 @@ func (k *klog) getMatchingPods() ([]string, error) {
// TODO: specify micro:service
// l["micro"] = "service"
if err := k.client.Get(r, l); err != nil {
if err := k.client.Get(r, client.GetLabels(l)); err != nil {
return nil, err
}

22
runtime/kubernetes/service.go
View File

@ -30,8 +30,8 @@ func newService(s *runtime.Service, c runtime.CreateOptions) *service {
name := client.Format(s.Name)
version := client.Format(s.Version)
kservice := client.NewService(name, version, c.Type)
kdeploy := client.NewDeployment(name, version, c.Type)
kservice := client.NewService(name, version, c.Type, c.Namespace)
kdeploy := client.NewDeployment(name, version, c.Type, c.Namespace)
// ensure the metadata is set
if kdeploy.Spec.Template.Metadata.Annotations == nil {
@ -115,9 +115,9 @@ func serviceResource(s *client.Service) *client.Resource {
}
// Start starts the Kubernetes service. It creates new kubernetes deployment and service API objects
func (s *service) Start(k client.Client) error {
func (s *service) Start(k client.Client, opts ...client.CreateOption) error {
// create deployment first; if we fail, we dont create service
if err := k.Create(deploymentResource(s.kdeploy)); err != nil {
if err := k.Create(deploymentResource(s.kdeploy), opts...); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime failed to create deployment: %v", err)
}
@ -129,7 +129,7 @@ func (s *service) Start(k client.Client) error {
return err
}
// create service now that the deployment has been created
if err := k.Create(serviceResource(s.kservice)); err != nil {
if err := k.Create(serviceResource(s.kservice), opts...); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime failed to create service: %v", err)
}
@ -146,9 +146,9 @@ func (s *service) Start(k client.Client) error {
return nil
}
func (s *service) Stop(k client.Client) error {
func (s *service) Stop(k client.Client, opts ...client.DeleteOption) error {
// first attempt to delete service
if err := k.Delete(serviceResource(s.kservice)); err != nil {
if err := k.Delete(serviceResource(s.kservice), opts...); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime failed to delete service: %v", err)
}
@ -156,7 +156,7 @@ func (s *service) Stop(k client.Client) error {
return err
}
// delete deployment once the service has been deleted
if err := k.Delete(deploymentResource(s.kdeploy)); err != nil {
if err := k.Delete(deploymentResource(s.kdeploy), opts...); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime failed to delete deployment: %v", err)
}
@ -169,15 +169,15 @@ func (s *service) Stop(k client.Client) error {
return nil
}
func (s *service) Update(k client.Client) error {
if err := k.Update(deploymentResource(s.kdeploy)); err != nil {
func (s *service) Update(k client.Client, opts ...client.UpdateOption) error {
if err := k.Update(deploymentResource(s.kdeploy), opts...); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime failed to update deployment: %v", err)
}
s.Status("error", err)
return err
}
if err := k.Update(serviceResource(s.kservice)); err != nil {
if err := k.Update(serviceResource(s.kservice), opts...); err != nil {
if logger.V(logger.DebugLevel, logger.DefaultLogger) {
logger.Debugf("Runtime failed to update service: %v", err)
}