Runtime refactoring and NetworkPolicy support (#2016)
This commit is contained in:
Prawn
GitHub
@@ -34,32 +34,52 @@ func (k *kubernetes) Init(opts ...runtime.Option) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (k *kubernetes) Logs(s *runtime.Service, options ...runtime.LogsOption) (runtime.Logs, error) {
|
||||
klo := newLog(k.client, s.Name, options...)
|
||||
func (k *kubernetes) Logs(resource runtime.Resource, options ...runtime.LogsOption) (runtime.Logs, error) {
|
||||
|
||||
if !klo.options.Stream {
|
||||
records, err := klo.Read()
|
||||
// Handle the various different types of resources:
|
||||
switch resource.Type() {
|
||||
case runtime.TypeNamespace:
|
||||
// noop (Namespace is not supported by *kubernetes.Logs())
|
||||
return nil, nil
|
||||
case runtime.TypeNetworkPolicy:
|
||||
// noop (NetworkPolicy is not supported by *kubernetes.Logs()))
|
||||
return nil, nil
|
||||
case runtime.TypeService:
|
||||
|
||||
// Assert the resource back into a *runtime.Service
|
||||
s, ok := resource.(*runtime.Service)
|
||||
if !ok {
|
||||
return nil, runtime.ErrInvalidResource
|
||||
}
|
||||
|
||||
klo := newLog(k.client, s.Name, options...)
|
||||
|
||||
if !klo.options.Stream {
|
||||
records, err := klo.Read()
|
||||
if err != nil {
|
||||
log.Errorf("Failed to get logs for service '%v' from k8s: %v", s.Name, err)
|
||||
return nil, err
|
||||
}
|
||||
kstream := &kubeStream{
|
||||
stream: make(chan runtime.Log),
|
||||
stop: make(chan bool),
|
||||
}
|
||||
go func() {
|
||||
for _, record := range records {
|
||||
kstream.Chan() <- record
|
||||
}
|
||||
kstream.Stop()
|
||||
}()
|
||||
return kstream, nil
|
||||
}
|
||||
stream, err := klo.Stream()
|
||||
if err != nil {
|
||||
log.Errorf("Failed to get logs for service '%v' from k8s: %v", s.Name, err)
|
||||
return nil, err
|
||||
}
|
||||
kstream := &kubeStream{
|
||||
stream: make(chan runtime.Log),
|
||||
stop: make(chan bool),
|
||||
}
|
||||
go func() {
|
||||
for _, record := range records {
|
||||
kstream.Chan() <- record
|
||||
}
|
||||
kstream.Stop()
|
||||
}()
|
||||
return kstream, nil
|
||||
return stream, nil
|
||||
default:
|
||||
return nil, runtime.ErrInvalidResource
|
||||
}
|
||||
stream, err := klo.Stream()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return stream, nil
|
||||
}
|
||||
|
||||
type kubeStream struct {
|
||||
@@ -92,11 +112,14 @@ func (k *kubeStream) Stop() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Creates a service
|
||||
func (k *kubernetes) Create(s *runtime.Service, opts ...runtime.CreateOption) error {
|
||||
// Create a resource
|
||||
func (k *kubernetes) Create(resource runtime.Resource, opts ...runtime.CreateOption) error {
|
||||
k.Lock()
|
||||
defer k.Unlock()
|
||||
return k.create(resource, opts...)
|
||||
}
|
||||
|
||||
func (k *kubernetes) create(resource runtime.Resource, opts ...runtime.CreateOption) error {
|
||||
// parse the options
|
||||
options := &runtime.CreateOptions{
|
||||
Type: k.options.Type,
|
||||
@@ -107,47 +130,74 @@ func (k *kubernetes) Create(s *runtime.Service, opts ...runtime.CreateOption) er
|
||||
o(options)
|
||||
}
|
||||
|
||||
// default the service's source and version
|
||||
if len(s.Source) == 0 {
|
||||
s.Source = k.options.Source
|
||||
}
|
||||
if len(s.Version) == 0 {
|
||||
s.Version = "latest"
|
||||
}
|
||||
|
||||
// ensure the namespace exists
|
||||
if err := k.ensureNamepaceExists(options.Namespace); err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// create a secret for the deployment
|
||||
if err := k.createCredentials(s, options); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// create the deployment
|
||||
if err := k.client.Create(client.NewDeployment(s, options), client.CreateNamespace(options.Namespace)); err != nil {
|
||||
if parseError(err).Reason == "AlreadyExists" {
|
||||
return runtime.ErrAlreadyExists
|
||||
// Handle the various different types of resources:
|
||||
switch resource.Type() {
|
||||
case runtime.TypeNamespace:
|
||||
// Assert the resource back into a *runtime.Namespace
|
||||
namespace, ok := resource.(*runtime.Namespace)
|
||||
if !ok {
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to create deployment: %v", err)
|
||||
return k.createNamespace(namespace)
|
||||
case runtime.TypeNetworkPolicy:
|
||||
// Assert the resource back into a *runtime.NetworkPolicy
|
||||
networkPolicy, ok := resource.(*runtime.NetworkPolicy)
|
||||
if !ok {
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
return err
|
||||
}
|
||||
return k.createNetworkPolicy(networkPolicy)
|
||||
case runtime.TypeService:
|
||||
|
||||
// create the service, one could already exist for another version so ignore ErrAlreadyExists
|
||||
if err := k.client.Create(client.NewService(s, options), client.CreateNamespace(options.Namespace)); err != nil {
|
||||
if parseError(err).Reason == "AlreadyExists" {
|
||||
// Assert the resource back into a *runtime.Service
|
||||
s, ok := resource.(*runtime.Service)
|
||||
if !ok {
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
|
||||
// default the service's source and version
|
||||
if len(s.Source) == 0 {
|
||||
s.Source = k.options.Source
|
||||
}
|
||||
if len(s.Version) == 0 {
|
||||
s.Version = "latest"
|
||||
}
|
||||
|
||||
// ensure the namespace exists
|
||||
if err := k.ensureNamepaceExists(options.Namespace); err != nil {
|
||||
return nil
|
||||
}
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to create service: %v", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
// create a secret for the deployment
|
||||
if err := k.createCredentials(s, options); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// create the deployment
|
||||
if err := k.client.Create(client.NewDeployment(s, options), client.CreateNamespace(options.Namespace)); err != nil {
|
||||
if parseError(err).Reason == "AlreadyExists" {
|
||||
return runtime.ErrAlreadyExists
|
||||
}
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to create deployment: %v", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// create the service, one could already exist for another version so ignore ErrAlreadyExists
|
||||
if err := k.client.Create(client.NewService(s, options), client.CreateNamespace(options.Namespace)); err != nil {
|
||||
if parseError(err).Reason == "AlreadyExists" {
|
||||
return nil
|
||||
}
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to create service: %v", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
default:
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
}
|
||||
|
||||
// Read returns all instances of given service
|
||||
@@ -180,8 +230,8 @@ func (k *kubernetes) Read(opts ...runtime.ReadOption) ([]*runtime.Service, error
|
||||
return k.getServices(client.GetNamespace(options.Namespace), client.GetLabels(labels))
|
||||
}
|
||||
|
||||
// Update the service in place
|
||||
func (k *kubernetes) Update(s *runtime.Service, opts ...runtime.UpdateOption) error {
|
||||
// Update a resource in place
|
||||
func (k *kubernetes) Update(resource runtime.Resource, opts ...runtime.UpdateOption) error {
|
||||
k.Lock()
|
||||
defer k.Unlock()
|
||||
|
||||
@@ -193,69 +243,91 @@ func (k *kubernetes) Update(s *runtime.Service, opts ...runtime.UpdateOption) er
|
||||
o(&options)
|
||||
}
|
||||
|
||||
// construct the query
|
||||
labels := map[string]string{}
|
||||
if len(s.Name) > 0 {
|
||||
labels["name"] = client.Format(s.Name)
|
||||
}
|
||||
if len(s.Version) > 0 {
|
||||
labels["version"] = client.Format(s.Version)
|
||||
}
|
||||
// Handle the various different types of resources:
|
||||
switch resource.Type() {
|
||||
case runtime.TypeNamespace:
|
||||
// noop (Namespace is not supported by *kubernetes.Update())
|
||||
return nil
|
||||
case runtime.TypeNetworkPolicy:
|
||||
// Assert the resource back into a *runtime.NetworkPolicy
|
||||
networkPolicy, ok := resource.(*runtime.NetworkPolicy)
|
||||
if !ok {
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
return k.updateNetworkPolicy(networkPolicy)
|
||||
case runtime.TypeService:
|
||||
|
||||
// get the existing deployments
|
||||
depList := new(client.DeploymentList)
|
||||
d := &client.Resource{
|
||||
Kind: "deployment",
|
||||
Value: depList,
|
||||
}
|
||||
depOpts := []client.GetOption{
|
||||
client.GetNamespace(options.Namespace),
|
||||
client.GetLabels(labels),
|
||||
}
|
||||
if err := k.client.Get(d, depOpts...); err != nil {
|
||||
return err
|
||||
} else if len(depList.Items) == 0 {
|
||||
return runtime.ErrNotFound
|
||||
}
|
||||
|
||||
// update the deployments which match the query
|
||||
for _, dep := range depList.Items {
|
||||
// the service wan't created by the k8s runtime
|
||||
if dep.Metadata == nil || dep.Metadata.Annotations == nil {
|
||||
continue
|
||||
// Assert the resource back into a *runtime.Service
|
||||
s, ok := resource.(*runtime.Service)
|
||||
if !ok {
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
|
||||
// update metadata
|
||||
for k, v := range s.Metadata {
|
||||
dep.Metadata.Annotations[k] = v
|
||||
// construct the query
|
||||
labels := map[string]string{}
|
||||
if len(s.Name) > 0 {
|
||||
labels["name"] = client.Format(s.Name)
|
||||
}
|
||||
if len(s.Version) > 0 {
|
||||
labels["version"] = client.Format(s.Version)
|
||||
}
|
||||
|
||||
// update build time annotation
|
||||
dep.Spec.Template.Metadata.Annotations["updated"] = fmt.Sprintf("%d", time.Now().Unix())
|
||||
|
||||
// update the deployment
|
||||
res := &client.Resource{
|
||||
// get the existing deployments
|
||||
depList := new(client.DeploymentList)
|
||||
d := &client.Resource{
|
||||
Kind: "deployment",
|
||||
Name: resourceName(s),
|
||||
Value: &dep,
|
||||
Value: depList,
|
||||
}
|
||||
if err := k.client.Update(res, client.UpdateNamespace(options.Namespace)); err != nil {
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to update deployment: %v", err)
|
||||
}
|
||||
depOpts := []client.GetOption{
|
||||
client.GetNamespace(options.Namespace),
|
||||
client.GetLabels(labels),
|
||||
}
|
||||
if err := k.client.Get(d, depOpts...); err != nil {
|
||||
return err
|
||||
} else if len(depList.Items) == 0 {
|
||||
return runtime.ErrNotFound
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
// update the deployments which match the query
|
||||
for _, dep := range depList.Items {
|
||||
// the service wan't created by the k8s runtime
|
||||
if dep.Metadata == nil || dep.Metadata.Annotations == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
// update metadata
|
||||
for k, v := range s.Metadata {
|
||||
dep.Metadata.Annotations[k] = v
|
||||
}
|
||||
|
||||
// update build time annotation
|
||||
dep.Spec.Template.Metadata.Annotations["updated"] = fmt.Sprintf("%d", time.Now().Unix())
|
||||
|
||||
// update the deployment
|
||||
res := &client.Resource{
|
||||
Kind: "deployment",
|
||||
Name: resourceName(s),
|
||||
Value: &dep,
|
||||
}
|
||||
if err := k.client.Update(res, client.UpdateNamespace(options.Namespace)); err != nil {
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to update deployment: %v", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
default:
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
}
|
||||
|
||||
// Delete removes a service
|
||||
func (k *kubernetes) Delete(s *runtime.Service, opts ...runtime.DeleteOption) error {
|
||||
// Delete removes a resource
|
||||
func (k *kubernetes) Delete(resource runtime.Resource, opts ...runtime.DeleteOption) error {
|
||||
k.Lock()
|
||||
defer k.Unlock()
|
||||
|
||||
// parse the options
|
||||
options := runtime.DeleteOptions{
|
||||
Namespace: client.DefaultNamespace,
|
||||
}
|
||||
@@ -263,51 +335,78 @@ func (k *kubernetes) Delete(s *runtime.Service, opts ...runtime.DeleteOption) er
|
||||
o(&options)
|
||||
}
|
||||
|
||||
// delete the deployment
|
||||
dep := client.NewDeployment(s, &runtime.CreateOptions{
|
||||
Type: k.options.Type,
|
||||
Namespace: options.Namespace,
|
||||
})
|
||||
if err := k.client.Delete(dep, client.DeleteNamespace(options.Namespace)); err != nil {
|
||||
if err == api.ErrNotFound {
|
||||
return runtime.ErrNotFound
|
||||
// Handle the various different types of resources:
|
||||
switch resource.Type() {
|
||||
case runtime.TypeNamespace:
|
||||
// Assert the resource back into a *runtime.Namespace
|
||||
namespace, ok := resource.(*runtime.Namespace)
|
||||
if !ok {
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to delete deployment: %v", err)
|
||||
return k.deleteNamespace(namespace)
|
||||
case runtime.TypeNetworkPolicy:
|
||||
// Assert the resource back into a *runtime.NetworkPolicy
|
||||
networkPolicy, ok := resource.(*runtime.NetworkPolicy)
|
||||
if !ok {
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
return err
|
||||
}
|
||||
return k.deleteNetworkPolicy(networkPolicy)
|
||||
case runtime.TypeService:
|
||||
|
||||
// delete the credentials
|
||||
if err := k.deleteCredentials(s, &runtime.CreateOptions{Namespace: options.Namespace}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// if there are more deployments for this service, then don't delete it
|
||||
labels := map[string]string{}
|
||||
if len(s.Name) > 0 {
|
||||
labels["name"] = client.Format(s.Name)
|
||||
}
|
||||
|
||||
// get the existing services. todo: refactor to just get the deployments
|
||||
services, err := k.getServices(client.GetNamespace(options.Namespace), client.GetLabels(labels))
|
||||
if err != nil || len(services) > 0 {
|
||||
return err
|
||||
}
|
||||
|
||||
// delete the service
|
||||
srv := client.NewService(s, &runtime.CreateOptions{
|
||||
Type: k.options.Type,
|
||||
Namespace: options.Namespace,
|
||||
})
|
||||
if err := k.client.Delete(srv, client.DeleteNamespace(options.Namespace)); err != nil {
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to delete service: %v", err)
|
||||
// Assert the resource back into a *runtime.Service
|
||||
s, ok := resource.(*runtime.Service)
|
||||
if !ok {
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
// delete the deployment
|
||||
dep := client.NewDeployment(s, &runtime.CreateOptions{
|
||||
Type: k.options.Type,
|
||||
Namespace: options.Namespace,
|
||||
})
|
||||
if err := k.client.Delete(dep, client.DeleteNamespace(options.Namespace)); err != nil {
|
||||
if err == api.ErrNotFound {
|
||||
return runtime.ErrNotFound
|
||||
}
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to delete deployment: %v", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// delete the credentials
|
||||
if err := k.deleteCredentials(s, &runtime.CreateOptions{Namespace: options.Namespace}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// if there are more deployments for this service, then don't delete it
|
||||
labels := map[string]string{}
|
||||
if len(s.Name) > 0 {
|
||||
labels["name"] = client.Format(s.Name)
|
||||
}
|
||||
|
||||
// get the existing services. todo: refactor to just get the deployments
|
||||
services, err := k.getServices(client.GetNamespace(options.Namespace), client.GetLabels(labels))
|
||||
if err != nil || len(services) > 0 {
|
||||
return err
|
||||
}
|
||||
|
||||
// delete the service
|
||||
srv := client.NewService(s, &runtime.CreateOptions{
|
||||
Type: k.options.Type,
|
||||
Namespace: options.Namespace,
|
||||
})
|
||||
if err := k.client.Delete(srv, client.DeleteNamespace(options.Namespace)); err != nil {
|
||||
if logger.V(logger.ErrorLevel, logger.DefaultLogger) {
|
||||
logger.Errorf("Runtime failed to delete service: %v", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
default:
|
||||
return runtime.ErrInvalidResource
|
||||
}
|
||||
}
|
||||
|
||||
// Start starts the runtime
|
||||
|
||||
Reference in New Issue
Block a user