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
))
307 timeout
: float = 1800,
309 db_dict
: dict = None,
310 kdu_name
: str = None,
311 namespace
: str = None,
315 :param cluster_uuid str: The UUID of the cluster to install to
316 :param kdu_model str: The name or path of a bundle to install
317 :param atomic bool: If set, waits until the model is active and resets
318 the cluster on failure.
319 :param timeout int: The time, in seconds, to wait for the install
321 :param params dict: Key-value pairs of instantiation parameters
322 :param kdu_name: Name of the KDU instance to be installed
323 :param namespace: K8s namespace to use for the KDU instance
325 :return: If successful, returns ?
330 raise K8sException("db_dict must be set")
332 raise K8sException("bundle must be set")
334 if bundle
.startswith("cs:"):
336 elif bundle
.startswith("http"):
340 new_workdir
= kdu_model
.strip(kdu_model
.split("/")[-1])
341 os
.chdir(new_workdir
)
342 bundle
= "local:{}".format(kdu_model
)
345 kdu_instance
= "{}-{}".format(kdu_name
, db_dict
["filter"]["_id"])
347 kdu_instance
= db_dict
["filter"]["_id"]
349 self
.log
.debug("Checking for model named {}".format(kdu_instance
))
351 # Create the new model
352 self
.log
.debug("Adding model: {}".format(kdu_instance
))
353 await self
.libjuju
.add_model(
354 model_name
=kdu_instance
,
355 cloud_name
=cluster_uuid
,
356 credential_name
=self
._get
_credential
_name
(cluster_uuid
),
360 # TODO: Instantiation parameters
363 "Juju bundle that models the KDU, in any of the following ways:
364 - <juju-repo>/<juju-bundle>
365 - <juju-bundle folder under k8s_models folder in the package>
366 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
368 - <URL_where_to_fetch_juju_bundle>
371 previous_workdir
= os
.getcwd()
372 except FileNotFoundError
:
373 previous_workdir
= "/app/storage"
375 self
.log
.debug("[install] deploying {}".format(bundle
))
376 await self
.libjuju
.deploy(
377 bundle
, model_name
=kdu_instance
, wait
=atomic
, timeout
=timeout
379 os
.chdir(previous_workdir
)
382 async def instances_list(self
, cluster_uuid
: str) -> list:
384 returns a list of deployed releases in a cluster
386 :param cluster_uuid: the cluster
395 kdu_model
: str = None,
400 :param cluster_uuid str: The UUID of the cluster to upgrade
401 :param kdu_instance str: The unique name of the KDU instance
402 :param kdu_model str: The name or path of the bundle to upgrade to
403 :param params dict: Key-value pairs of instantiation parameters
405 :return: If successful, reference to the new revision number of the
409 # TODO: Loop through the bundle and upgrade each charm individually
412 The API doesn't have a concept of bundle upgrades, because there are
413 many possible changes: charm revision, disk, number of units, etc.
415 As such, we are only supporting a limited subset of upgrades. We'll
416 upgrade the charm revision but leave storage and scale untouched.
418 Scale changes should happen through OSM constructs, and changes to
419 storage would require a redeployment of the service, at least in this
422 raise MethodNotImplemented()
434 :param cluster_uuid str: The UUID of the cluster to rollback
435 :param kdu_instance str: The unique name of the KDU instance
436 :param revision int: The revision to revert to. If omitted, rolls back
437 the previous upgrade.
439 :return: If successful, returns the revision of active KDU instance,
440 or raises an exception
442 raise MethodNotImplemented()
446 async def uninstall(self
, cluster_uuid
: str, kdu_instance
: str) -> bool:
447 """Uninstall a KDU instance
449 :param cluster_uuid str: The UUID of the cluster
450 :param kdu_instance str: The unique name of the KDU instance
452 :return: Returns True if successful, or raises an exception
455 self
.log
.debug("[uninstall] Destroying model")
457 await self
.libjuju
.destroy_model(kdu_instance
, total_timeout
=3600)
459 # self.log.debug("[uninstall] Model destroyed and disconnecting")
460 # await controller.disconnect()
463 # TODO: Remove these commented lines
464 # if not self.authenticated:
465 # self.log.debug("[uninstall] Connecting to controller")
466 # await self.login(cluster_uuid)
468 async def exec_primitive(
470 cluster_uuid
: str = None,
471 kdu_instance
: str = None,
472 primitive_name
: str = None,
473 timeout
: float = 300,
475 db_dict
: dict = None,
477 """Exec primitive (Juju action)
479 :param cluster_uuid str: The UUID of the cluster
480 :param kdu_instance str: The unique name of the KDU instance
481 :param primitive_name: Name of action that will be executed
482 :param timeout: Timeout for action execution
483 :param params: Dictionary of all the parameters needed for the action
484 :db_dict: Dictionary for any additional data
486 :return: Returns the output of the action
489 if not params
or "application-name" not in params
:
491 "Missing application-name argument, \
492 argument needed for K8s actions"
496 "[exec_primitive] Getting model "
497 "kdu_instance: {}".format(kdu_instance
)
499 application_name
= params
["application-name"]
500 actions
= await self
.libjuju
.get_actions(application_name
, kdu_instance
)
501 if primitive_name
not in actions
:
502 raise K8sException("Primitive {} not found".format(primitive_name
))
503 output
, status
= await self
.libjuju
.execute_action(
504 application_name
, kdu_instance
, primitive_name
, **params
507 if status
!= "completed":
509 "status is not completed: {} output: {}".format(status
, output
)
514 except Exception as e
:
515 error_msg
= "Error executing primitive {}: {}".format(primitive_name
, e
)
516 self
.log
.error(error_msg
)
517 raise K8sException(message
=error_msg
)
521 async def inspect_kdu(
527 Inspects a bundle and returns a dictionary of config parameters and
528 their default values.
530 :param kdu_model str: The name or path of the bundle to inspect.
532 :return: If successful, returns a dictionary of available parameters
533 and their default values.
537 if not os
.path
.exists(kdu_model
):
538 raise K8sException("file {} not found".format(kdu_model
))
540 with
open(kdu_model
, "r") as f
:
541 bundle
= yaml
.safe_load(f
.read())
545 'description': 'Test bundle',
546 'bundle': 'kubernetes',
549 'charm': 'cs:~charmed-osm/mariadb-k8s-20',
552 'password': 'manopw',
553 'root_password': 'osm4u',
556 'series': 'kubernetes'
561 # TODO: This should be returned in an agreed-upon format
562 kdu
= bundle
["applications"]
572 If available, returns the README of the bundle.
574 :param kdu_model str: The name or path of a bundle
576 :return: If found, returns the contents of the README.
580 files
= ["README", "README.txt", "README.md"]
581 path
= os
.path
.dirname(kdu_model
)
582 for file in os
.listdir(path
):
584 with
open(file, "r") as f
:
590 async def status_kdu(
595 """Get the status of the KDU
597 Get the current status of the KDU instance.
599 :param cluster_uuid str: The UUID of the cluster
600 :param kdu_instance str: The unique id of the KDU instance
602 :return: Returns a dictionary containing namespace, state, resources,
606 model_status
= await self
.libjuju
.get_model_status(kdu_instance
)
607 for name
in model_status
.applications
:
608 application
= model_status
.applications
[name
]
609 status
[name
] = {"status": application
["status"]["status"]}
613 async def get_services(
614 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
616 """Return a list of services of a kdu_instance"""
618 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
620 kubecfg
= tempfile
.NamedTemporaryFile()
621 with
open(kubecfg
.name
, "w") as kubecfg_file
:
622 kubecfg_file
.write(credentials
)
623 kubectl
= Kubectl(config_file
=kubecfg
.name
)
625 return kubectl
.get_services(
626 field_selector
="metadata.namespace={}".format(kdu_instance
)
629 async def get_service(
630 self
, cluster_uuid
: str, service_name
: str, namespace
: str
632 """Return data for a specific service inside a namespace"""
634 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
636 kubecfg
= tempfile
.NamedTemporaryFile()
637 with
open(kubecfg
.name
, "w") as kubecfg_file
:
638 kubecfg_file
.write(credentials
)
639 kubectl
= Kubectl(config_file
=kubecfg
.name
)
641 return kubectl
.get_services(
642 field_selector
="metadata.name={},metadata.namespace={}".format(
643 service_name
, namespace
647 def get_credentials(self
, cluster_uuid
: str) -> str:
649 Get Cluster Kubeconfig
651 k8scluster
= self
.db
.get_one(
652 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
655 self
.db
.encrypt_decrypt_fields(
656 k8scluster
.get("credentials"),
658 ["password", "secret"],
659 schema_version
=k8scluster
["schema_version"],
660 salt
=k8scluster
["_id"],
663 return yaml
.safe_dump(k8scluster
.get("credentials"))
665 def _get_credential_name(self
, cluster_uuid
: str) -> str:
667 Get credential name for a k8s cloud
669 We cannot use the cluster_uuid for the credential name directly,
670 because it cannot start with a number, it must start with a letter.
671 Therefore, the k8s cloud credential name will be "cred-" followed
674 :param: cluster_uuid: Cluster UUID of the kubernetes cloud (=cloud_name)
676 :return: Name to use for the credential name.
678 return "cred-{}".format(cluster_uuid
)
684 """Get the namespace UUID
685 Gets the namespace's unique name
687 :param cluster_uuid str: The UUID of the cluster
688 :returns: The namespace UUID, or raises an exception
692 def _create_cluster_role(
696 labels
: Dict
[str, str],
698 cluster_roles
= kubectl
.clients
[RBAC_CLIENT
].list_cluster_role(
699 field_selector
="metadata.name={}".format(name
)
702 if len(cluster_roles
.items
) > 0:
704 "Cluster role with metadata.name={} already exists".format(name
)
707 metadata
= V1ObjectMeta(name
=name
, labels
=labels
, namespace
=ADMIN_NAMESPACE
)
709 cluster_role
= V1ClusterRole(
712 V1PolicyRule(api_groups
=["*"], resources
=["*"], verbs
=["*"]),
713 V1PolicyRule(non_resource_ur_ls
=["*"], verbs
=["*"]),
717 kubectl
.clients
[RBAC_CLIENT
].create_cluster_role(cluster_role
)
719 def _delete_cluster_role(self
, kubectl
: Kubectl
, name
: str):
720 kubectl
.clients
[RBAC_CLIENT
].delete_cluster_role(name
)
722 def _create_service_account(
726 labels
: Dict
[str, str],
728 service_accounts
= kubectl
.clients
[CORE_CLIENT
].list_namespaced_service_account(
729 ADMIN_NAMESPACE
, field_selector
="metadata.name={}".format(name
)
731 if len(service_accounts
.items
) > 0:
733 "Service account with metadata.name={} already exists".format(name
)
736 metadata
= V1ObjectMeta(name
=name
, labels
=labels
, namespace
=ADMIN_NAMESPACE
)
737 service_account
= V1ServiceAccount(metadata
=metadata
)
739 kubectl
.clients
[CORE_CLIENT
].create_namespaced_service_account(
740 ADMIN_NAMESPACE
, service_account
743 def _delete_service_account(self
, kubectl
: Kubectl
, name
: str):
744 kubectl
.clients
[CORE_CLIENT
].delete_namespaced_service_account(
745 name
, ADMIN_NAMESPACE
748 def _create_cluster_role_binding(
752 labels
: Dict
[str, str],
754 role_bindings
= kubectl
.clients
[RBAC_CLIENT
].list_cluster_role_binding(
755 field_selector
="metadata.name={}".format(name
)
757 if len(role_bindings
.items
) > 0:
758 raise Exception("Generated rbac id already exists")
760 role_binding
= V1ClusterRoleBinding(
761 metadata
=V1ObjectMeta(name
=name
, labels
=labels
),
762 role_ref
=V1RoleRef(kind
="ClusterRole", name
=name
, api_group
=""),
764 V1Subject(kind
="ServiceAccount", name
=name
, namespace
=ADMIN_NAMESPACE
)
767 kubectl
.clients
[RBAC_CLIENT
].create_cluster_role_binding(role_binding
)
769 def _delete_cluster_role_binding(self
, kubectl
: Kubectl
, name
: str):
770 kubectl
.clients
[RBAC_CLIENT
].delete_cluster_role_binding(name
)
772 async def _get_secret_data(self
, kubectl
: Kubectl
, name
: str) -> (str, str):
773 v1_core
= kubectl
.clients
[CORE_CLIENT
]
779 service_accounts
= v1_core
.list_namespaced_service_account(
780 ADMIN_NAMESPACE
, field_selector
="metadata.name={}".format(name
)
782 if len(service_accounts
.items
) == 0:
784 "Service account not found with metadata.name={}".format(name
)
786 service_account
= service_accounts
.items
[0]
787 if service_account
.secrets
and len(service_account
.secrets
) > 0:
788 secret_name
= service_account
.secrets
[0].name
789 if secret_name
is not None or not retries_limit
:
793 "Failed getting the secret from service account {}".format(name
)
795 secret
= v1_core
.list_namespaced_secret(
797 field_selector
="metadata.name={}".format(secret_name
),
800 token
= secret
.data
[SERVICE_ACCOUNT_TOKEN_KEY
]
801 client_certificate_data
= secret
.data
[SERVICE_ACCOUNT_ROOT_CA_KEY
]
804 base64
.b64decode(token
).decode("utf-8"),
805 base64
.b64decode(client_certificate_data
).decode("utf-8"),