1 # Copyright 2019 Canonical Ltd.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
23 from n2vc
.config
import ModelConfig
24 from n2vc
.exceptions
import K8sException
, N2VCBadArgumentsException
25 from n2vc
.k8s_conn
import K8sConnector
26 from n2vc
.kubectl
import Kubectl
, CORE_CLIENT
, RBAC_CLIENT
27 from .exceptions
import MethodNotImplemented
28 from n2vc
.utils
import base64_to_cacert
29 from n2vc
.libjuju
import Libjuju
30 from n2vc
.utils
import obj_to_dict
, obj_to_yaml
32 from kubernetes
.client
.models
import (
42 from typing
import Dict
44 SERVICE_ACCOUNT_TOKEN_KEY
= "token"
45 SERVICE_ACCOUNT_ROOT_CA_KEY
= "ca.crt"
46 RBAC_LABEL_KEY_NAME
= "rbac-id"
48 ADMIN_NAMESPACE
= "kube-system"
49 RBAC_STACK_PREFIX
= "juju-credential"
52 def generate_rbac_id():
53 return binascii
.hexlify(os
.urandom(4)).decode()
56 class K8sJujuConnector(K8sConnector
):
61 kubectl_command
: str = "/usr/bin/kubectl",
62 juju_command
: str = "/usr/bin/juju",
66 vca_config
: dict = None,
69 :param fs: file system for kubernetes and helm configuration
70 :param db: Database object
71 :param kubectl_command: path to kubectl executable
72 :param helm_command: path to helm executable
74 :param: loop: Asyncio loop
78 K8sConnector
.__init
__(
82 on_update_db
=on_update_db
,
86 self
.loop
= loop
or asyncio
.get_event_loop()
87 self
.log
.debug("Initializing K8S Juju connector")
89 required_vca_config
= [
95 if not vca_config
or not all(k
in vca_config
for k
in required_vca_config
):
96 raise N2VCBadArgumentsException(
97 message
="Missing arguments in vca_config: {}".format(vca_config
),
98 bad_args
=required_vca_config
,
100 port
= vca_config
["port"] if "port" in vca_config
else 17070
101 url
= "{}:{}".format(vca_config
["host"], port
)
102 model_config
= ModelConfig(vca_config
)
103 username
= vca_config
["user"]
104 secret
= vca_config
["secret"]
105 ca_cert
= base64_to_cacert(vca_config
["ca_cert"])
107 self
.libjuju
= Libjuju(
109 api_proxy
=None, # Not needed for k8s charms
110 model_config
=model_config
,
118 self
.log
.debug("K8S Juju connector initialized")
119 # TODO: Remove these commented lines:
120 # self.authenticated = False
122 # self.juju_secret = ""
129 namespace
: str = "kube-system",
130 reuse_cluster_uuid
: str = None,
133 It prepares a given K8s cluster environment to run Juju bundles.
135 :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid
137 :param namespace: optional namespace to be used for juju. By default,
138 'kube-system' will be used
139 :param reuse_cluster_uuid: existing cluster uuid for reuse
140 :return: uuid of the K8s cluster and True if connector has installed some
141 software in the cluster
142 (on error, an exception will be raised)
145 cluster_uuid
= reuse_cluster_uuid
or str(uuid
.uuid4())
147 kubecfg
= tempfile
.NamedTemporaryFile()
148 with
open(kubecfg
.name
, "w") as kubecfg_file
:
149 kubecfg_file
.write(k8s_creds
)
150 kubectl
= Kubectl(config_file
=kubecfg
.name
)
152 # CREATING RESOURCES IN K8S
153 rbac_id
= generate_rbac_id()
154 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
155 labels
= {RBAC_STACK_PREFIX
: rbac_id
}
157 # Create cleanup dictionary to clean up created resources
158 # if it fails in the middle of the process
161 self
._create
_cluster
_role
(
168 "delete": self
._delete
_cluster
_role
,
169 "args": (kubectl
, metadata_name
),
173 self
._create
_service
_account
(
180 "delete": self
._delete
_service
_account
,
181 "args": (kubectl
, metadata_name
),
185 self
._create
_cluster
_role
_binding
(
192 "delete": self
._delete
_service
_account
,
193 "args": (kubectl
, metadata_name
),
196 token
, client_cert_data
= await self
._get
_secret
_data
(
201 default_storage_class
= kubectl
.get_default_storage_class()
202 await self
.libjuju
.add_k8s(
206 client_cert_data
=client_cert_data
,
207 configuration
=kubectl
.configuration
,
208 storage_class
=default_storage_class
,
209 credential_name
=self
._get
_credential
_name
(cluster_uuid
),
211 return cluster_uuid
, True
212 except Exception as e
:
213 self
.log
.error("Error initializing k8scluster: {}".format(e
))
214 if len(cleanup_data
) > 0:
215 self
.log
.debug("Cleaning up created resources in k8s cluster...")
216 for item
in cleanup_data
:
217 delete_function
= item
["delete"]
218 delete_args
= item
["args"]
219 delete_function(*delete_args
)
220 self
.log
.debug("Cleanup finished")
223 """Repo Management"""
229 _type
: str = "charm",
231 raise MethodNotImplemented()
233 async def repo_list(self
):
234 raise MethodNotImplemented()
236 async def repo_remove(
240 raise MethodNotImplemented()
242 async def synchronize_repos(self
, cluster_uuid
: str, name
: str):
244 Returns None as currently add_repo is not implemented
251 self
, cluster_uuid
: str, force
: bool = False, uninstall_sw
: bool = False
255 Resets the Kubernetes cluster by removing the model that represents it.
257 :param cluster_uuid str: The UUID of the cluster to reset
258 :return: Returns True if successful or raises an exception.
262 self
.log
.debug("[reset] Removing k8s cloud")
264 cloud_creds
= await self
.libjuju
.get_cloud_credentials(
266 self
._get
_credential
_name
(cluster_uuid
),
269 await self
.libjuju
.remove_cloud(cluster_uuid
)
271 kubecfg
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
273 kubecfg_file
= tempfile
.NamedTemporaryFile()
274 with
open(kubecfg_file
.name
, "w") as f
:
276 kubectl
= Kubectl(config_file
=kubecfg_file
.name
)
279 self
._delete
_cluster
_role
_binding
,
280 self
._delete
_service
_account
,
281 self
._delete
_cluster
_role
,
284 credential_attrs
= cloud_creds
[0].result
["attrs"]
285 if RBAC_LABEL_KEY_NAME
in credential_attrs
:
286 rbac_id
= credential_attrs
[RBAC_LABEL_KEY_NAME
]
287 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
288 delete_args
= (kubectl
, metadata_name
)
289 for delete_func
in delete_functions
:
291 delete_func(*delete_args
)
292 except Exception as e
:
293 self
.log
.warning("Cannot remove resource in K8s {}".format(e
))
295 except Exception as e
:
296 self
.log
.debug("Caught exception during reset: {}".format(e
))
308 timeout
: float = 1800,
310 db_dict
: dict = None,
311 kdu_name
: str = None,
312 namespace
: str = None,
316 :param cluster_uuid str: The UUID of the cluster to install to
317 :param kdu_model str: The name or path of a bundle to install
318 :param kdu_instance: Kdu instance name
319 :param atomic bool: If set, waits until the model is active and resets
320 the cluster on failure.
321 :param timeout int: The time, in seconds, to wait for the install
323 :param params dict: Key-value pairs of instantiation parameters
324 :param kdu_name: Name of the KDU instance to be installed
325 :param namespace: K8s namespace to use for the KDU instance
327 :return: If successful, returns ?
332 raise K8sException("db_dict must be set")
334 raise K8sException("bundle must be set")
336 if bundle
.startswith("cs:"):
338 elif bundle
.startswith("http"):
342 new_workdir
= kdu_model
.strip(kdu_model
.split("/")[-1])
343 os
.chdir(new_workdir
)
344 bundle
= "local:{}".format(kdu_model
)
346 self
.log
.debug("Checking for model named {}".format(kdu_instance
))
348 # Create the new model
349 self
.log
.debug("Adding model: {}".format(kdu_instance
))
350 await self
.libjuju
.add_model(
351 model_name
=kdu_instance
,
352 cloud_name
=cluster_uuid
,
353 credential_name
=self
._get
_credential
_name
(cluster_uuid
),
357 # TODO: Instantiation parameters
360 "Juju bundle that models the KDU, in any of the following ways:
361 - <juju-repo>/<juju-bundle>
362 - <juju-bundle folder under k8s_models folder in the package>
363 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
365 - <URL_where_to_fetch_juju_bundle>
368 previous_workdir
= os
.getcwd()
369 except FileNotFoundError
:
370 previous_workdir
= "/app/storage"
372 self
.log
.debug("[install] deploying {}".format(bundle
))
373 await self
.libjuju
.deploy(
374 bundle
, model_name
=kdu_instance
, wait
=atomic
, timeout
=timeout
376 os
.chdir(previous_workdir
)
377 if self
.on_update_db
:
378 await self
.on_update_db(cluster_uuid
, kdu_instance
, filter=db_dict
["filter"])
381 async def instances_list(self
, cluster_uuid
: str) -> list:
383 returns a list of deployed releases in a cluster
385 :param cluster_uuid: the cluster
394 kdu_model
: str = None,
399 :param cluster_uuid str: The UUID of the cluster to upgrade
400 :param kdu_instance str: The unique name of the KDU instance
401 :param kdu_model str: The name or path of the bundle to upgrade to
402 :param params dict: Key-value pairs of instantiation parameters
404 :return: If successful, reference to the new revision number of the
408 # TODO: Loop through the bundle and upgrade each charm individually
411 The API doesn't have a concept of bundle upgrades, because there are
412 many possible changes: charm revision, disk, number of units, etc.
414 As such, we are only supporting a limited subset of upgrades. We'll
415 upgrade the charm revision but leave storage and scale untouched.
417 Scale changes should happen through OSM constructs, and changes to
418 storage would require a redeployment of the service, at least in this
421 raise MethodNotImplemented()
433 :param cluster_uuid str: The UUID of the cluster to rollback
434 :param kdu_instance str: The unique name of the KDU instance
435 :param revision int: The revision to revert to. If omitted, rolls back
436 the previous upgrade.
438 :return: If successful, returns the revision of active KDU instance,
439 or raises an exception
441 raise MethodNotImplemented()
445 async def uninstall(self
, cluster_uuid
: str, kdu_instance
: str) -> bool:
446 """Uninstall a KDU instance
448 :param cluster_uuid str: The UUID of the cluster
449 :param kdu_instance str: The unique name of the KDU instance
451 :return: Returns True if successful, or raises an exception
454 self
.log
.debug("[uninstall] Destroying model")
456 await self
.libjuju
.destroy_model(kdu_instance
, total_timeout
=3600)
458 # self.log.debug("[uninstall] Model destroyed and disconnecting")
459 # await controller.disconnect()
462 # TODO: Remove these commented lines
463 # if not self.authenticated:
464 # self.log.debug("[uninstall] Connecting to controller")
465 # await self.login(cluster_uuid)
467 async def exec_primitive(
469 cluster_uuid
: str = None,
470 kdu_instance
: str = None,
471 primitive_name
: str = None,
472 timeout
: float = 300,
474 db_dict
: dict = None,
476 """Exec primitive (Juju action)
478 :param cluster_uuid str: The UUID of the cluster
479 :param kdu_instance str: The unique name of the KDU instance
480 :param primitive_name: Name of action that will be executed
481 :param timeout: Timeout for action execution
482 :param params: Dictionary of all the parameters needed for the action
483 :db_dict: Dictionary for any additional data
485 :return: Returns the output of the action
488 if not params
or "application-name" not in params
:
490 "Missing application-name argument, \
491 argument needed for K8s actions"
495 "[exec_primitive] Getting model "
496 "kdu_instance: {}".format(kdu_instance
)
498 application_name
= params
["application-name"]
499 actions
= await self
.libjuju
.get_actions(application_name
, kdu_instance
)
500 if primitive_name
not in actions
:
501 raise K8sException("Primitive {} not found".format(primitive_name
))
502 output
, status
= await self
.libjuju
.execute_action(
503 application_name
, kdu_instance
, primitive_name
, **params
506 if status
!= "completed":
508 "status is not completed: {} output: {}".format(status
, output
)
510 if self
.on_update_db
:
511 await self
.on_update_db(cluster_uuid
, kdu_instance
, filter=db_dict
["filter"])
515 except Exception as e
:
516 error_msg
= "Error executing primitive {}: {}".format(primitive_name
, e
)
517 self
.log
.error(error_msg
)
518 raise K8sException(message
=error_msg
)
522 async def inspect_kdu(
528 Inspects a bundle and returns a dictionary of config parameters and
529 their default values.
531 :param kdu_model str: The name or path of the bundle to inspect.
533 :return: If successful, returns a dictionary of available parameters
534 and their default values.
538 if not os
.path
.exists(kdu_model
):
539 raise K8sException("file {} not found".format(kdu_model
))
541 with
open(kdu_model
, "r") as f
:
542 bundle
= yaml
.safe_load(f
.read())
546 'description': 'Test bundle',
547 'bundle': 'kubernetes',
550 'charm': 'cs:~charmed-osm/mariadb-k8s-20',
553 'password': 'manopw',
554 'root_password': 'osm4u',
557 'series': 'kubernetes'
562 # TODO: This should be returned in an agreed-upon format
563 kdu
= bundle
["applications"]
573 If available, returns the README of the bundle.
575 :param kdu_model str: The name or path of a bundle
577 :return: If found, returns the contents of the README.
581 files
= ["README", "README.txt", "README.md"]
582 path
= os
.path
.dirname(kdu_model
)
583 for file in os
.listdir(path
):
585 with
open(file, "r") as f
:
591 async def status_kdu(
595 complete_status
: bool = False,
596 yaml_format
: bool = False
598 """Get the status of the KDU
600 Get the current status of the KDU instance.
602 :param cluster_uuid str: The UUID of the cluster
603 :param kdu_instance str: The unique id of the KDU instance
604 :param complete_status: To get the complete_status of the KDU
605 :param yaml_format: To get the status in proper format for NSR record
607 :return: Returns a dictionary containing namespace, state, resources,
608 and deployment_time and returns complete_status if complete_status is True
612 model_status
= await self
.libjuju
.get_model_status(kdu_instance
)
614 if not complete_status
:
615 for name
in model_status
.applications
:
616 application
= model_status
.applications
[name
]
617 status
[name
] = {"status": application
["status"]["status"]}
620 return obj_to_yaml(model_status
)
622 return obj_to_dict(model_status
)
626 async def update_vca_status(self
, vcastatus
: dict, kdu_instance
: str):
628 Add all configs, actions, executed actions of all applications in a model to vcastatus dict
630 :param vcastatus dict: dict containing vcastatus
631 :param kdu_instance str: The unique id of the KDU instance
636 for model_name
in vcastatus
:
637 # Adding executed actions
638 vcastatus
[model_name
]["executedActions"] = \
639 await self
.libjuju
.get_executed_actions(kdu_instance
)
641 for application
in vcastatus
[model_name
]["applications"]:
642 # Adding application actions
643 vcastatus
[model_name
]["applications"][application
]["actions"] = \
644 await self
.libjuju
.get_actions(application
, kdu_instance
)
645 # Adding application configs
646 vcastatus
[model_name
]["applications"][application
]["configs"] = \
647 await self
.libjuju
.get_application_configs(kdu_instance
, application
)
649 except Exception as e
:
650 self
.log
.debug("Error in updating vca status: {}".format(str(e
)))
652 async def get_services(
653 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
655 """Return a list of services of a kdu_instance"""
657 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
659 kubecfg
= tempfile
.NamedTemporaryFile()
660 with
open(kubecfg
.name
, "w") as kubecfg_file
:
661 kubecfg_file
.write(credentials
)
662 kubectl
= Kubectl(config_file
=kubecfg
.name
)
664 return kubectl
.get_services(
665 field_selector
="metadata.namespace={}".format(kdu_instance
)
668 async def get_service(
669 self
, cluster_uuid
: str, service_name
: str, namespace
: str
671 """Return data for a specific service inside a namespace"""
673 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
675 kubecfg
= tempfile
.NamedTemporaryFile()
676 with
open(kubecfg
.name
, "w") as kubecfg_file
:
677 kubecfg_file
.write(credentials
)
678 kubectl
= Kubectl(config_file
=kubecfg
.name
)
680 return kubectl
.get_services(
681 field_selector
="metadata.name={},metadata.namespace={}".format(
682 service_name
, namespace
686 def get_credentials(self
, cluster_uuid
: str) -> str:
688 Get Cluster Kubeconfig
690 k8scluster
= self
.db
.get_one(
691 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
694 self
.db
.encrypt_decrypt_fields(
695 k8scluster
.get("credentials"),
697 ["password", "secret"],
698 schema_version
=k8scluster
["schema_version"],
699 salt
=k8scluster
["_id"],
702 return yaml
.safe_dump(k8scluster
.get("credentials"))
704 def _get_credential_name(self
, cluster_uuid
: str) -> str:
706 Get credential name for a k8s cloud
708 We cannot use the cluster_uuid for the credential name directly,
709 because it cannot start with a number, it must start with a letter.
710 Therefore, the k8s cloud credential name will be "cred-" followed
713 :param: cluster_uuid: Cluster UUID of the kubernetes cloud (=cloud_name)
715 :return: Name to use for the credential name.
717 return "cred-{}".format(cluster_uuid
)
723 """Get the namespace UUID
724 Gets the namespace's unique name
726 :param cluster_uuid str: The UUID of the cluster
727 :returns: The namespace UUID, or raises an exception
731 def _create_cluster_role(
735 labels
: Dict
[str, str],
737 cluster_roles
= kubectl
.clients
[RBAC_CLIENT
].list_cluster_role(
738 field_selector
="metadata.name={}".format(name
)
741 if len(cluster_roles
.items
) > 0:
743 "Cluster role with metadata.name={} already exists".format(name
)
746 metadata
= V1ObjectMeta(name
=name
, labels
=labels
, namespace
=ADMIN_NAMESPACE
)
748 cluster_role
= V1ClusterRole(
751 V1PolicyRule(api_groups
=["*"], resources
=["*"], verbs
=["*"]),
752 V1PolicyRule(non_resource_ur_ls
=["*"], verbs
=["*"]),
756 kubectl
.clients
[RBAC_CLIENT
].create_cluster_role(cluster_role
)
758 def _delete_cluster_role(self
, kubectl
: Kubectl
, name
: str):
759 kubectl
.clients
[RBAC_CLIENT
].delete_cluster_role(name
)
761 def _create_service_account(
765 labels
: Dict
[str, str],
767 service_accounts
= kubectl
.clients
[CORE_CLIENT
].list_namespaced_service_account(
768 ADMIN_NAMESPACE
, field_selector
="metadata.name={}".format(name
)
770 if len(service_accounts
.items
) > 0:
772 "Service account with metadata.name={} already exists".format(name
)
775 metadata
= V1ObjectMeta(name
=name
, labels
=labels
, namespace
=ADMIN_NAMESPACE
)
776 service_account
= V1ServiceAccount(metadata
=metadata
)
778 kubectl
.clients
[CORE_CLIENT
].create_namespaced_service_account(
779 ADMIN_NAMESPACE
, service_account
782 def _delete_service_account(self
, kubectl
: Kubectl
, name
: str):
783 kubectl
.clients
[CORE_CLIENT
].delete_namespaced_service_account(
784 name
, ADMIN_NAMESPACE
787 def _create_cluster_role_binding(
791 labels
: Dict
[str, str],
793 role_bindings
= kubectl
.clients
[RBAC_CLIENT
].list_cluster_role_binding(
794 field_selector
="metadata.name={}".format(name
)
796 if len(role_bindings
.items
) > 0:
797 raise Exception("Generated rbac id already exists")
799 role_binding
= V1ClusterRoleBinding(
800 metadata
=V1ObjectMeta(name
=name
, labels
=labels
),
801 role_ref
=V1RoleRef(kind
="ClusterRole", name
=name
, api_group
=""),
803 V1Subject(kind
="ServiceAccount", name
=name
, namespace
=ADMIN_NAMESPACE
)
806 kubectl
.clients
[RBAC_CLIENT
].create_cluster_role_binding(role_binding
)
808 def _delete_cluster_role_binding(self
, kubectl
: Kubectl
, name
: str):
809 kubectl
.clients
[RBAC_CLIENT
].delete_cluster_role_binding(name
)
811 async def _get_secret_data(self
, kubectl
: Kubectl
, name
: str) -> (str, str):
812 v1_core
= kubectl
.clients
[CORE_CLIENT
]
818 service_accounts
= v1_core
.list_namespaced_service_account(
819 ADMIN_NAMESPACE
, field_selector
="metadata.name={}".format(name
)
821 if len(service_accounts
.items
) == 0:
823 "Service account not found with metadata.name={}".format(name
)
825 service_account
= service_accounts
.items
[0]
826 if service_account
.secrets
and len(service_account
.secrets
) > 0:
827 secret_name
= service_account
.secrets
[0].name
828 if secret_name
is not None or not retries_limit
:
832 "Failed getting the secret from service account {}".format(name
)
834 secret
= v1_core
.list_namespaced_secret(
836 field_selector
="metadata.name={}".format(secret_name
),
839 token
= secret
.data
[SERVICE_ACCOUNT_TOKEN_KEY
]
840 client_certificate_data
= secret
.data
[SERVICE_ACCOUNT_ROOT_CA_KEY
]
843 base64
.b64decode(token
).decode("utf-8"),
844 base64
.b64decode(client_certificate_data
).decode("utf-8"),
848 def generate_kdu_instance_name(**kwargs
):
849 db_dict
= kwargs
.get("db_dict")
850 kdu_name
= kwargs
.get("kdu_name", None)
852 kdu_instance
= "{}-{}".format(kdu_name
, db_dict
["filter"]["_id"])
854 kdu_instance
= db_dict
["filter"]["_id"]