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
.exceptions
import K8sException
, N2VCBadArgumentsException
24 from n2vc
.k8s_conn
import K8sConnector
25 from n2vc
.kubectl
import Kubectl
, CORE_CLIENT
, RBAC_CLIENT
26 from .exceptions
import MethodNotImplemented
27 from n2vc
.utils
import base64_to_cacert
28 from n2vc
.libjuju
import Libjuju
30 from kubernetes
.client
.models
import (
40 from typing
import Dict
42 SERVICE_ACCOUNT_TOKEN_KEY
= "token"
43 SERVICE_ACCOUNT_ROOT_CA_KEY
= "ca.crt"
44 RBAC_LABEL_KEY_NAME
= "rbac-id"
46 ADMIN_NAMESPACE
= "kube-system"
47 RBAC_STACK_PREFIX
= "juju-credential"
50 def generate_rbac_id():
51 return binascii
.hexlify(os
.urandom(4)).decode()
54 class K8sJujuConnector(K8sConnector
):
59 kubectl_command
: str = "/usr/bin/kubectl",
60 juju_command
: str = "/usr/bin/juju",
64 vca_config
: dict = None,
67 :param fs: file system for kubernetes and helm configuration
68 :param db: Database object
69 :param kubectl_command: path to kubectl executable
70 :param helm_command: path to helm executable
72 :param: loop: Asyncio loop
76 K8sConnector
.__init
__(
80 on_update_db
=on_update_db
,
84 self
.loop
= loop
or asyncio
.get_event_loop()
85 self
.log
.debug("Initializing K8S Juju connector")
87 required_vca_config
= [
93 if not vca_config
or not all(k
in vca_config
for k
in required_vca_config
):
94 raise N2VCBadArgumentsException(
95 message
="Missing arguments in vca_config: {}".format(vca_config
),
96 bad_args
=required_vca_config
,
98 port
= vca_config
["port"] if "port" in vca_config
else 17070
99 url
= "{}:{}".format(vca_config
["host"], port
)
100 enable_os_upgrade
= vca_config
.get("enable_os_upgrade", True)
101 apt_mirror
= vca_config
.get("apt_mirror", None)
102 username
= vca_config
["user"]
103 secret
= vca_config
["secret"]
104 ca_cert
= base64_to_cacert(vca_config
["ca_cert"])
106 self
.libjuju
= Libjuju(
108 api_proxy
=None, # Not needed for k8s charms
109 enable_os_upgrade
=enable_os_upgrade
,
110 apt_mirror
=apt_mirror
,
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
)
379 async def instances_list(self
, cluster_uuid
: str) -> list:
381 returns a list of deployed releases in a cluster
383 :param cluster_uuid: the cluster
392 kdu_model
: str = None,
397 :param cluster_uuid str: The UUID of the cluster to upgrade
398 :param kdu_instance str: The unique name of the KDU instance
399 :param kdu_model str: The name or path of the bundle to upgrade to
400 :param params dict: Key-value pairs of instantiation parameters
402 :return: If successful, reference to the new revision number of the
406 # TODO: Loop through the bundle and upgrade each charm individually
409 The API doesn't have a concept of bundle upgrades, because there are
410 many possible changes: charm revision, disk, number of units, etc.
412 As such, we are only supporting a limited subset of upgrades. We'll
413 upgrade the charm revision but leave storage and scale untouched.
415 Scale changes should happen through OSM constructs, and changes to
416 storage would require a redeployment of the service, at least in this
419 raise MethodNotImplemented()
431 :param cluster_uuid str: The UUID of the cluster to rollback
432 :param kdu_instance str: The unique name of the KDU instance
433 :param revision int: The revision to revert to. If omitted, rolls back
434 the previous upgrade.
436 :return: If successful, returns the revision of active KDU instance,
437 or raises an exception
439 raise MethodNotImplemented()
443 async def uninstall(self
, cluster_uuid
: str, kdu_instance
: str) -> bool:
444 """Uninstall a KDU instance
446 :param cluster_uuid str: The UUID of the cluster
447 :param kdu_instance str: The unique name of the KDU instance
449 :return: Returns True if successful, or raises an exception
452 self
.log
.debug("[uninstall] Destroying model")
454 await self
.libjuju
.destroy_model(kdu_instance
, total_timeout
=3600)
456 # self.log.debug("[uninstall] Model destroyed and disconnecting")
457 # await controller.disconnect()
460 # TODO: Remove these commented lines
461 # if not self.authenticated:
462 # self.log.debug("[uninstall] Connecting to controller")
463 # await self.login(cluster_uuid)
465 async def exec_primitive(
467 cluster_uuid
: str = None,
468 kdu_instance
: str = None,
469 primitive_name
: str = None,
470 timeout
: float = 300,
472 db_dict
: dict = None,
474 """Exec primitive (Juju action)
476 :param cluster_uuid str: The UUID of the cluster
477 :param kdu_instance str: The unique name of the KDU instance
478 :param primitive_name: Name of action that will be executed
479 :param timeout: Timeout for action execution
480 :param params: Dictionary of all the parameters needed for the action
481 :db_dict: Dictionary for any additional data
483 :return: Returns the output of the action
486 if not params
or "application-name" not in params
:
488 "Missing application-name argument, \
489 argument needed for K8s actions"
493 "[exec_primitive] Getting model "
494 "kdu_instance: {}".format(kdu_instance
)
496 application_name
= params
["application-name"]
497 actions
= await self
.libjuju
.get_actions(application_name
, kdu_instance
)
498 if primitive_name
not in actions
:
499 raise K8sException("Primitive {} not found".format(primitive_name
))
500 output
, status
= await self
.libjuju
.execute_action(
501 application_name
, kdu_instance
, primitive_name
, **params
504 if status
!= "completed":
506 "status is not completed: {} output: {}".format(status
, output
)
511 except Exception as e
:
512 error_msg
= "Error executing primitive {}: {}".format(primitive_name
, e
)
513 self
.log
.error(error_msg
)
514 raise K8sException(message
=error_msg
)
518 async def inspect_kdu(
524 Inspects a bundle and returns a dictionary of config parameters and
525 their default values.
527 :param kdu_model str: The name or path of the bundle to inspect.
529 :return: If successful, returns a dictionary of available parameters
530 and their default values.
534 if not os
.path
.exists(kdu_model
):
535 raise K8sException("file {} not found".format(kdu_model
))
537 with
open(kdu_model
, "r") as f
:
538 bundle
= yaml
.safe_load(f
.read())
542 'description': 'Test bundle',
543 'bundle': 'kubernetes',
546 'charm': 'cs:~charmed-osm/mariadb-k8s-20',
549 'password': 'manopw',
550 'root_password': 'osm4u',
553 'series': 'kubernetes'
558 # TODO: This should be returned in an agreed-upon format
559 kdu
= bundle
["applications"]
569 If available, returns the README of the bundle.
571 :param kdu_model str: The name or path of a bundle
573 :return: If found, returns the contents of the README.
577 files
= ["README", "README.txt", "README.md"]
578 path
= os
.path
.dirname(kdu_model
)
579 for file in os
.listdir(path
):
581 with
open(file, "r") as f
:
587 async def status_kdu(
592 """Get the status of the KDU
594 Get the current status of the KDU instance.
596 :param cluster_uuid str: The UUID of the cluster
597 :param kdu_instance str: The unique id of the KDU instance
599 :return: Returns a dictionary containing namespace, state, resources,
603 model_status
= await self
.libjuju
.get_model_status(kdu_instance
)
604 for name
in model_status
.applications
:
605 application
= model_status
.applications
[name
]
606 status
[name
] = {"status": application
["status"]["status"]}
610 async def get_services(
611 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
613 """Return a list of services of a kdu_instance"""
615 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
617 kubecfg
= tempfile
.NamedTemporaryFile()
618 with
open(kubecfg
.name
, "w") as kubecfg_file
:
619 kubecfg_file
.write(credentials
)
620 kubectl
= Kubectl(config_file
=kubecfg
.name
)
622 return kubectl
.get_services(
623 field_selector
="metadata.namespace={}".format(kdu_instance
)
626 async def get_service(
627 self
, cluster_uuid
: str, service_name
: str, namespace
: str
629 """Return data for a specific service inside a namespace"""
631 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
633 kubecfg
= tempfile
.NamedTemporaryFile()
634 with
open(kubecfg
.name
, "w") as kubecfg_file
:
635 kubecfg_file
.write(credentials
)
636 kubectl
= Kubectl(config_file
=kubecfg
.name
)
638 return kubectl
.get_services(
639 field_selector
="metadata.name={},metadata.namespace={}".format(
640 service_name
, namespace
644 def get_credentials(self
, cluster_uuid
: str) -> str:
646 Get Cluster Kubeconfig
648 k8scluster
= self
.db
.get_one(
649 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
652 self
.db
.encrypt_decrypt_fields(
653 k8scluster
.get("credentials"),
655 ["password", "secret"],
656 schema_version
=k8scluster
["schema_version"],
657 salt
=k8scluster
["_id"],
660 return yaml
.safe_dump(k8scluster
.get("credentials"))
662 def _get_credential_name(self
, cluster_uuid
: str) -> str:
664 Get credential name for a k8s cloud
666 We cannot use the cluster_uuid for the credential name directly,
667 because it cannot start with a number, it must start with a letter.
668 Therefore, the k8s cloud credential name will be "cred-" followed
671 :param: cluster_uuid: Cluster UUID of the kubernetes cloud (=cloud_name)
673 :return: Name to use for the credential name.
675 return "cred-{}".format(cluster_uuid
)
681 """Get the namespace UUID
682 Gets the namespace's unique name
684 :param cluster_uuid str: The UUID of the cluster
685 :returns: The namespace UUID, or raises an exception
689 def _create_cluster_role(
693 labels
: Dict
[str, str],
695 cluster_roles
= kubectl
.clients
[RBAC_CLIENT
].list_cluster_role(
696 field_selector
="metadata.name={}".format(name
)
699 if len(cluster_roles
.items
) > 0:
701 "Cluster role with metadata.name={} already exists".format(name
)
704 metadata
= V1ObjectMeta(name
=name
, labels
=labels
, namespace
=ADMIN_NAMESPACE
)
706 cluster_role
= V1ClusterRole(
709 V1PolicyRule(api_groups
=["*"], resources
=["*"], verbs
=["*"]),
710 V1PolicyRule(non_resource_ur_ls
=["*"], verbs
=["*"]),
714 kubectl
.clients
[RBAC_CLIENT
].create_cluster_role(cluster_role
)
716 def _delete_cluster_role(self
, kubectl
: Kubectl
, name
: str):
717 kubectl
.clients
[RBAC_CLIENT
].delete_cluster_role(name
)
719 def _create_service_account(
723 labels
: Dict
[str, str],
725 service_accounts
= kubectl
.clients
[CORE_CLIENT
].list_namespaced_service_account(
726 ADMIN_NAMESPACE
, field_selector
="metadata.name={}".format(name
)
728 if len(service_accounts
.items
) > 0:
730 "Service account with metadata.name={} already exists".format(name
)
733 metadata
= V1ObjectMeta(name
=name
, labels
=labels
, namespace
=ADMIN_NAMESPACE
)
734 service_account
= V1ServiceAccount(metadata
=metadata
)
736 kubectl
.clients
[CORE_CLIENT
].create_namespaced_service_account(
737 ADMIN_NAMESPACE
, service_account
740 def _delete_service_account(self
, kubectl
: Kubectl
, name
: str):
741 kubectl
.clients
[CORE_CLIENT
].delete_namespaced_service_account(
742 name
, ADMIN_NAMESPACE
745 def _create_cluster_role_binding(
749 labels
: Dict
[str, str],
751 role_bindings
= kubectl
.clients
[RBAC_CLIENT
].list_cluster_role_binding(
752 field_selector
="metadata.name={}".format(name
)
754 if len(role_bindings
.items
) > 0:
755 raise Exception("Generated rbac id already exists")
757 role_binding
= V1ClusterRoleBinding(
758 metadata
=V1ObjectMeta(name
=name
, labels
=labels
),
759 role_ref
=V1RoleRef(kind
="ClusterRole", name
=name
, api_group
=""),
761 V1Subject(kind
="ServiceAccount", name
=name
, namespace
=ADMIN_NAMESPACE
)
764 kubectl
.clients
[RBAC_CLIENT
].create_cluster_role_binding(role_binding
)
766 def _delete_cluster_role_binding(self
, kubectl
: Kubectl
, name
: str):
767 kubectl
.clients
[RBAC_CLIENT
].delete_cluster_role_binding(name
)
769 async def _get_secret_data(self
, kubectl
: Kubectl
, name
: str) -> (str, str):
770 v1_core
= kubectl
.clients
[CORE_CLIENT
]
776 service_accounts
= v1_core
.list_namespaced_service_account(
777 ADMIN_NAMESPACE
, field_selector
="metadata.name={}".format(name
)
779 if len(service_accounts
.items
) == 0:
781 "Service account not found with metadata.name={}".format(name
)
783 service_account
= service_accounts
.items
[0]
784 if service_account
.secrets
and len(service_account
.secrets
) > 0:
785 secret_name
= service_account
.secrets
[0].name
786 if secret_name
is not None or not retries_limit
:
790 "Failed getting the secret from service account {}".format(name
)
792 secret
= v1_core
.list_namespaced_secret(
794 field_selector
="metadata.name={}".format(secret_name
),
797 token
= secret
.data
[SERVICE_ACCOUNT_TOKEN_KEY
]
798 client_certificate_data
= secret
.data
[SERVICE_ACCOUNT_ROOT_CA_KEY
]
801 base64
.b64decode(token
).decode("utf-8"),
802 base64
.b64decode(client_certificate_data
).decode("utf-8"),
806 def generate_kdu_instance_name(**kwargs
):
807 db_dict
= kwargs
.get("db_dict")
808 kdu_name
= kwargs
.get("kdu_name", None)
810 kdu_instance
= "{}-{}".format(kdu_name
, db_dict
["filter"]["_id"])
812 kdu_instance
= db_dict
["filter"]["_id"]