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.
16 from typing
import Union
23 from n2vc
.config
import EnvironConfig
24 from n2vc
.exceptions
import K8sException
25 from n2vc
.k8s_conn
import K8sConnector
26 from n2vc
.kubectl
import Kubectl
27 from .exceptions
import MethodNotImplemented
28 from n2vc
.libjuju
import Libjuju
29 from n2vc
.utils
import obj_to_dict
, obj_to_yaml
30 from n2vc
.store
import MotorStore
31 from n2vc
.vca
.cloud
import Cloud
32 from n2vc
.vca
.connection
import get_connection
35 RBAC_LABEL_KEY_NAME
= "rbac-id"
36 RBAC_STACK_PREFIX
= "juju-credential"
39 def generate_rbac_id():
40 return binascii
.hexlify(os
.urandom(4)).decode()
43 class K8sJujuConnector(K8sConnector
):
50 kubectl_command
: str = "/usr/bin/kubectl",
51 juju_command
: str = "/usr/bin/juju",
57 :param fs: file system for kubernetes and helm configuration
58 :param db: Database object
59 :param kubectl_command: path to kubectl executable
60 :param helm_command: path to helm executable
62 :param: loop: Asyncio loop
66 K8sConnector
.__init
__(
70 on_update_db
=on_update_db
,
74 self
.loop
= loop
or asyncio
.get_event_loop()
75 self
.log
.debug("Initializing K8S Juju connector")
77 db_uri
= EnvironConfig(prefixes
=["OSMLCM_", "OSMMON_"]).get("database_uri")
78 self
._store
= MotorStore(db_uri
)
79 self
.loading_libjuju
= asyncio
.Lock(loop
=self
.loop
)
80 self
.uninstall_locks
= {}
82 self
.log
.debug("K8S Juju connector initialized")
83 # TODO: Remove these commented lines:
84 # self.authenticated = False
86 # self.juju_secret = ""
93 namespace
: str = "kube-system",
94 reuse_cluster_uuid
: str = None,
98 It prepares a given K8s cluster environment to run Juju bundles.
100 :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid
102 :param namespace: optional namespace to be used for juju. By default,
103 'kube-system' will be used
104 :param reuse_cluster_uuid: existing cluster uuid for reuse
105 :param: kwargs: Additional parameters
108 :return: uuid of the K8s cluster and True if connector has installed some
109 software in the cluster
110 (on error, an exception will be raised)
112 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
114 cluster_uuid
= reuse_cluster_uuid
or str(uuid
.uuid4())
115 kubectl
= self
._get
_kubectl
(k8s_creds
)
117 # CREATING RESOURCES IN K8S
118 rbac_id
= generate_rbac_id()
119 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
120 labels
= {RBAC_STACK_PREFIX
: rbac_id
}
122 # Create cleanup dictionary to clean up created resources
123 # if it fails in the middle of the process
126 kubectl
.create_cluster_role(
132 "delete": kubectl
.delete_cluster_role
,
133 "args": (metadata_name
,),
137 kubectl
.create_service_account(
143 "delete": kubectl
.delete_service_account
,
144 "args": (metadata_name
,),
148 kubectl
.create_cluster_role_binding(
154 "delete": kubectl
.delete_service_account
,
155 "args": (metadata_name
,),
158 token
, client_cert_data
= await kubectl
.get_secret_data(
162 default_storage_class
= kubectl
.get_default_storage_class()
163 await libjuju
.add_k8s(
167 client_cert_data
=client_cert_data
,
168 configuration
=kubectl
.configuration
,
169 storage_class
=default_storage_class
,
170 credential_name
=self
._get
_credential
_name
(cluster_uuid
),
172 return cluster_uuid
, True
173 except Exception as e
:
174 self
.log
.error("Error initializing k8scluster: {}".format(e
))
175 if len(cleanup_data
) > 0:
176 self
.log
.debug("Cleaning up created resources in k8s cluster...")
177 for item
in cleanup_data
:
178 delete_function
= item
["delete"]
179 delete_args
= item
["args"]
180 delete_function(*delete_args
)
181 self
.log
.debug("Cleanup finished")
184 """Repo Management"""
190 _type
: str = "charm",
192 raise MethodNotImplemented()
194 async def repo_list(self
):
195 raise MethodNotImplemented()
197 async def repo_remove(
201 raise MethodNotImplemented()
203 async def synchronize_repos(self
, cluster_uuid
: str, name
: str):
205 Returns None as currently add_repo is not implemented
215 uninstall_sw
: bool = False,
220 Resets the Kubernetes cluster by removing the model that represents it.
222 :param cluster_uuid str: The UUID of the cluster to reset
223 :param force: Force reset
224 :param uninstall_sw: Boolean to uninstall sw
225 :param: kwargs: Additional parameters
228 :return: Returns True if successful or raises an exception.
232 self
.log
.debug("[reset] Removing k8s cloud")
233 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
235 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
237 cloud_creds
= await libjuju
.get_cloud_credentials(cloud
)
239 await libjuju
.remove_cloud(cluster_uuid
)
241 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
243 kubectl
= self
._get
_kubectl
(credentials
)
246 kubectl
.delete_cluster_role_binding
,
247 kubectl
.delete_service_account
,
248 kubectl
.delete_cluster_role
,
251 credential_attrs
= cloud_creds
[0].result
["attrs"]
252 if RBAC_LABEL_KEY_NAME
in credential_attrs
:
253 rbac_id
= credential_attrs
[RBAC_LABEL_KEY_NAME
]
254 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
255 for delete_func
in delete_functions
:
257 delete_func(metadata_name
)
258 except Exception as e
:
259 self
.log
.warning("Cannot remove resource in K8s {}".format(e
))
261 except Exception as e
:
262 self
.log
.debug("Caught exception during reset: {}".format(e
))
274 timeout
: float = 1800,
276 db_dict
: dict = None,
277 kdu_name
: str = None,
278 namespace
: str = None,
283 :param cluster_uuid str: The UUID of the cluster to install to
284 :param kdu_model str: The name or path of a bundle to install
285 :param kdu_instance: Kdu instance name
286 :param atomic bool: If set, waits until the model is active and resets
287 the cluster on failure.
288 :param timeout int: The time, in seconds, to wait for the install
290 :param params dict: Key-value pairs of instantiation parameters
291 :param kdu_name: Name of the KDU instance to be installed
292 :param namespace: K8s namespace to use for the KDU instance
293 :param kwargs: Additional parameters
296 :return: If successful, returns ?
298 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
302 raise K8sException("db_dict must be set")
304 raise K8sException("bundle must be set")
306 if bundle
.startswith("cs:"):
307 # For Juju Bundles provided by the Charm Store
309 elif bundle
.startswith("ch:"):
310 # For Juju Bundles provided by the Charm Hub (this only works for juju version >= 2.9)
312 elif bundle
.startswith("http"):
316 new_workdir
= kdu_model
.strip(kdu_model
.split("/")[-1])
317 os
.chdir(new_workdir
)
318 bundle
= "local:{}".format(kdu_model
)
320 self
.log
.debug("Checking for model named {}".format(kdu_instance
))
322 # Create the new model
323 self
.log
.debug("Adding model: {}".format(kdu_instance
))
324 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
325 await libjuju
.add_model(kdu_instance
, cloud
)
328 # TODO: Instantiation parameters
331 "Juju bundle that models the KDU, in any of the following ways:
332 - <juju-repo>/<juju-bundle>
333 - <juju-bundle folder under k8s_models folder in the package>
334 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
336 - <URL_where_to_fetch_juju_bundle>
339 previous_workdir
= os
.getcwd()
340 except FileNotFoundError
:
341 previous_workdir
= "/app/storage"
343 self
.log
.debug("[install] deploying {}".format(bundle
))
344 await libjuju
.deploy(
345 bundle
, model_name
=kdu_instance
, wait
=atomic
, timeout
=timeout
347 os
.chdir(previous_workdir
)
348 if self
.on_update_db
:
349 await self
.on_update_db(
352 filter=db_dict
["filter"],
353 vca_id
=kwargs
.get("vca_id"),
362 total_timeout
: float = 1800,
365 """Scale an application in a model
367 :param: kdu_instance str: KDU instance name
368 :param: scale int: Scale to which to set this application
369 :param: resource_name str: Resource name (Application name)
370 :param: timeout float: The time, in seconds, to wait for the install
372 :param kwargs: Additional parameters
375 :return: If successful, returns True
379 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
380 await libjuju
.scale_application(
381 model_name
=kdu_instance
,
382 application_name
=resource_name
,
384 total_timeout
=total_timeout
,
386 except Exception as e
:
387 error_msg
= "Error scaling application {} in kdu instance {}: {}".format(
388 resource_name
, kdu_instance
, e
390 self
.log
.error(error_msg
)
391 raise K8sException(message
=error_msg
)
394 async def get_scale_count(
400 """Get an application scale count
402 :param: resource_name str: Resource name (Application name)
403 :param: kdu_instance str: KDU instance name
404 :param kwargs: Additional parameters
406 :return: Return application instance count
409 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
410 status
= await libjuju
.get_model_status(kdu_instance
)
411 return len(status
.applications
[resource_name
].units
)
412 except Exception as e
:
413 error_msg
= "Error getting scale count from application {} in kdu instance {}: {}".format(
414 resource_name
, kdu_instance
, e
416 self
.log
.error(error_msg
)
417 raise K8sException(message
=error_msg
)
419 async def instances_list(self
, cluster_uuid
: str) -> list:
421 returns a list of deployed releases in a cluster
423 :param cluster_uuid: the cluster
432 kdu_model
: str = None,
437 :param cluster_uuid str: The UUID of the cluster to upgrade
438 :param kdu_instance str: The unique name of the KDU instance
439 :param kdu_model str: The name or path of the bundle to upgrade to
440 :param params dict: Key-value pairs of instantiation parameters
442 :return: If successful, reference to the new revision number of the
446 # TODO: Loop through the bundle and upgrade each charm individually
449 The API doesn't have a concept of bundle upgrades, because there are
450 many possible changes: charm revision, disk, number of units, etc.
452 As such, we are only supporting a limited subset of upgrades. We'll
453 upgrade the charm revision but leave storage and scale untouched.
455 Scale changes should happen through OSM constructs, and changes to
456 storage would require a redeployment of the service, at least in this
459 raise MethodNotImplemented()
471 :param cluster_uuid str: The UUID of the cluster to rollback
472 :param kdu_instance str: The unique name of the KDU instance
473 :param revision int: The revision to revert to. If omitted, rolls back
474 the previous upgrade.
476 :return: If successful, returns the revision of active KDU instance,
477 or raises an exception
479 raise MethodNotImplemented()
489 """Uninstall a KDU instance
491 :param cluster_uuid str: The UUID of the cluster
492 :param kdu_instance str: The unique name of the KDU instance
493 :param kwargs: Additional parameters
496 :return: Returns True if successful, or raises an exception
499 self
.log
.debug("[uninstall] Destroying model")
501 will_not_delete
= False
502 if kdu_instance
not in self
.uninstall_locks
:
503 self
.uninstall_locks
[kdu_instance
] = asyncio
.Lock(loop
=self
.loop
)
504 delete_lock
= self
.uninstall_locks
[kdu_instance
]
506 while delete_lock
.locked():
507 will_not_delete
= True
508 await asyncio
.sleep(0.1)
511 self
.log
.info("Model {} deleted by another worker.".format(kdu_instance
))
515 async with delete_lock
:
516 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
518 await libjuju
.destroy_model(kdu_instance
, total_timeout
=3600)
520 self
.uninstall_locks
.pop(kdu_instance
)
522 self
.log
.debug(f
"[uninstall] Model {kdu_instance} destroyed")
525 async def exec_primitive(
527 cluster_uuid
: str = None,
528 kdu_instance
: str = None,
529 primitive_name
: str = None,
530 timeout
: float = 300,
532 db_dict
: dict = None,
535 """Exec primitive (Juju action)
537 :param cluster_uuid str: The UUID of the cluster
538 :param kdu_instance str: The unique name of the KDU instance
539 :param primitive_name: Name of action that will be executed
540 :param timeout: Timeout for action execution
541 :param params: Dictionary of all the parameters needed for the action
542 :param db_dict: Dictionary for any additional data
543 :param kwargs: Additional parameters
546 :return: Returns the output of the action
548 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
550 if not params
or "application-name" not in params
:
552 "Missing application-name argument, \
553 argument needed for K8s actions"
557 "[exec_primitive] Getting model "
558 "kdu_instance: {}".format(kdu_instance
)
560 application_name
= params
["application-name"]
561 actions
= await libjuju
.get_actions(application_name
, kdu_instance
)
562 if primitive_name
not in actions
:
563 raise K8sException("Primitive {} not found".format(primitive_name
))
564 output
, status
= await libjuju
.execute_action(
565 application_name
, kdu_instance
, primitive_name
, **params
568 if status
!= "completed":
570 "status is not completed: {} output: {}".format(status
, output
)
572 if self
.on_update_db
:
573 await self
.on_update_db(
574 cluster_uuid
, kdu_instance
, filter=db_dict
["filter"]
579 except Exception as e
:
580 error_msg
= "Error executing primitive {}: {}".format(primitive_name
, e
)
581 self
.log
.error(error_msg
)
582 raise K8sException(message
=error_msg
)
586 async def inspect_kdu(
592 Inspects a bundle and returns a dictionary of config parameters and
593 their default values.
595 :param kdu_model str: The name or path of the bundle to inspect.
597 :return: If successful, returns a dictionary of available parameters
598 and their default values.
602 if not os
.path
.exists(kdu_model
):
603 raise K8sException("file {} not found".format(kdu_model
))
605 with
open(kdu_model
, "r") as f
:
606 bundle
= yaml
.safe_load(f
.read())
610 'description': 'Test bundle',
611 'bundle': 'kubernetes',
614 'charm': 'cs:~charmed-osm/mariadb-k8s-20',
617 'password': 'manopw',
618 'root_password': 'osm4u',
621 'series': 'kubernetes'
626 # TODO: This should be returned in an agreed-upon format
627 kdu
= bundle
["applications"]
637 If available, returns the README of the bundle.
639 :param kdu_model str: The name or path of a bundle
641 :return: If found, returns the contents of the README.
645 files
= ["README", "README.txt", "README.md"]
646 path
= os
.path
.dirname(kdu_model
)
647 for file in os
.listdir(path
):
649 with
open(file, "r") as f
:
655 async def status_kdu(
659 complete_status
: bool = False,
660 yaml_format
: bool = False,
662 ) -> Union
[str, dict]:
663 """Get the status of the KDU
665 Get the current status of the KDU instance.
667 :param cluster_uuid str: The UUID of the cluster
668 :param kdu_instance str: The unique id of the KDU instance
669 :param complete_status: To get the complete_status of the KDU
670 :param yaml_format: To get the status in proper format for NSR record
671 :param: kwargs: Additional parameters
674 :return: Returns a dictionary containing namespace, state, resources,
675 and deployment_time and returns complete_status if complete_status is True
677 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
680 model_status
= await libjuju
.get_model_status(kdu_instance
)
682 if not complete_status
:
683 for name
in model_status
.applications
:
684 application
= model_status
.applications
[name
]
685 status
[name
] = {"status": application
["status"]["status"]}
688 return obj_to_yaml(model_status
)
690 return obj_to_dict(model_status
)
694 async def update_vca_status(self
, vcastatus
: dict, kdu_instance
: str, **kwargs
):
696 Add all configs, actions, executed actions of all applications in a model to vcastatus dict
698 :param vcastatus dict: dict containing vcastatus
699 :param kdu_instance str: The unique id of the KDU instance
700 :param: kwargs: Additional parameters
705 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
707 for model_name
in vcastatus
:
708 # Adding executed actions
709 vcastatus
[model_name
][
711 ] = await libjuju
.get_executed_actions(kdu_instance
)
713 for application
in vcastatus
[model_name
]["applications"]:
714 # Adding application actions
715 vcastatus
[model_name
]["applications"][application
][
717 ] = await libjuju
.get_actions(application
, kdu_instance
)
718 # Adding application configs
719 vcastatus
[model_name
]["applications"][application
][
721 ] = await libjuju
.get_application_configs(kdu_instance
, application
)
723 except Exception as e
:
724 self
.log
.debug("Error in updating vca status: {}".format(str(e
)))
726 async def get_services(
727 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
729 """Return a list of services of a kdu_instance"""
731 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
732 kubectl
= self
._get
_kubectl
(credentials
)
733 return kubectl
.get_services(
734 field_selector
="metadata.namespace={}".format(kdu_instance
)
737 async def get_service(
738 self
, cluster_uuid
: str, service_name
: str, namespace
: str
740 """Return data for a specific service inside a namespace"""
742 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
743 kubectl
= self
._get
_kubectl
(credentials
)
744 return kubectl
.get_services(
745 field_selector
="metadata.name={},metadata.namespace={}".format(
746 service_name
, namespace
750 def get_credentials(self
, cluster_uuid
: str) -> str:
752 Get Cluster Kubeconfig
754 k8scluster
= self
.db
.get_one(
755 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
758 self
.db
.encrypt_decrypt_fields(
759 k8scluster
.get("credentials"),
761 ["password", "secret"],
762 schema_version
=k8scluster
["schema_version"],
763 salt
=k8scluster
["_id"],
766 return yaml
.safe_dump(k8scluster
.get("credentials"))
768 def _get_credential_name(self
, cluster_uuid
: str) -> str:
770 Get credential name for a k8s cloud
772 We cannot use the cluster_uuid for the credential name directly,
773 because it cannot start with a number, it must start with a letter.
774 Therefore, the k8s cloud credential name will be "cred-" followed
777 :param: cluster_uuid: Cluster UUID of the kubernetes cloud (=cloud_name)
779 :return: Name to use for the credential name.
781 return "cred-{}".format(cluster_uuid
)
787 """Get the namespace UUID
788 Gets the namespace's unique name
790 :param cluster_uuid str: The UUID of the cluster
791 :returns: The namespace UUID, or raises an exception
796 def generate_kdu_instance_name(**kwargs
):
797 db_dict
= kwargs
.get("db_dict")
798 kdu_name
= kwargs
.get("kdu_name", None)
800 kdu_instance
= "{}-{}".format(kdu_name
, db_dict
["filter"]["_id"])
802 kdu_instance
= db_dict
["filter"]["_id"]
805 async def _get_libjuju(self
, vca_id
: str = None) -> Libjuju
:
809 :param: vca_id: VCA ID
810 If None, get a libjuju object with a Connection to the default VCA
811 Else, geta libjuju object with a Connection to the specified VCA
814 while self
.loading_libjuju
.locked():
815 await asyncio
.sleep(0.1)
817 async with self
.loading_libjuju
:
818 vca_connection
= await get_connection(self
._store
)
819 self
.libjuju
= Libjuju(vca_connection
, loop
=self
.loop
, log
=self
.log
)
822 vca_connection
= await get_connection(self
._store
, vca_id
)
830 def _get_kubectl(self
, credentials
: str) -> Kubectl
:
834 :param: kubeconfig_credentials: Kubeconfig credentials
836 kubecfg
= tempfile
.NamedTemporaryFile()
837 with
open(kubecfg
.name
, "w") as kubecfg_file
:
838 kubecfg_file
.write(credentials
)
839 return Kubectl(config_file
=kubecfg
.name
)