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.
22 from n2vc
.config
import EnvironConfig
23 from n2vc
.definitions
import RelationEndpoint
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
)
81 self
.log
.debug("K8S Juju connector initialized")
82 # TODO: Remove these commented lines:
83 # self.authenticated = False
85 # self.juju_secret = ""
92 namespace
: str = "kube-system",
93 reuse_cluster_uuid
: str = None,
97 It prepares a given K8s cluster environment to run Juju bundles.
99 :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid
101 :param namespace: optional namespace to be used for juju. By default,
102 'kube-system' will be used
103 :param reuse_cluster_uuid: existing cluster uuid for reuse
104 :param: kwargs: Additional parameters
107 :return: uuid of the K8s cluster and True if connector has installed some
108 software in the cluster
109 (on error, an exception will be raised)
111 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
113 cluster_uuid
= reuse_cluster_uuid
or str(uuid
.uuid4())
114 kubectl
= self
._get
_kubectl
(k8s_creds
)
116 # CREATING RESOURCES IN K8S
117 rbac_id
= generate_rbac_id()
118 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
119 labels
= {RBAC_STACK_PREFIX
: rbac_id
}
121 # Create cleanup dictionary to clean up created resources
122 # if it fails in the middle of the process
125 self
.log
.debug("Initializing K8s cluster for juju")
126 kubectl
.create_cluster_role(
130 self
.log
.debug("Cluster role created")
133 "delete": kubectl
.delete_cluster_role
,
134 "args": (metadata_name
,),
138 kubectl
.create_service_account(
142 self
.log
.debug("Service account created")
145 "delete": kubectl
.delete_service_account
,
146 "args": (metadata_name
,),
150 kubectl
.create_cluster_role_binding(
154 self
.log
.debug("Role binding created")
157 "delete": kubectl
.delete_service_account
,
158 "args": (metadata_name
,),
161 token
, client_cert_data
= await kubectl
.get_secret_data(
165 default_storage_class
= kubectl
.get_default_storage_class()
166 self
.log
.debug("Default storage class: {}".format(default_storage_class
))
167 await libjuju
.add_k8s(
171 client_cert_data
=client_cert_data
,
172 configuration
=kubectl
.configuration
,
173 storage_class
=default_storage_class
,
174 credential_name
=self
._get
_credential
_name
(cluster_uuid
),
176 self
.log
.debug("K8s cluster added to juju controller")
177 return cluster_uuid
, True
178 except Exception as e
:
179 self
.log
.error("Error initializing k8scluster: {}".format(e
), exc_info
=True)
180 if len(cleanup_data
) > 0:
181 self
.log
.debug("Cleaning up created resources in k8s cluster...")
182 for item
in cleanup_data
:
183 delete_function
= item
["delete"]
184 delete_args
= item
["args"]
185 delete_function(*delete_args
)
186 self
.log
.debug("Cleanup finished")
189 """Repo Management"""
195 _type
: str = "charm",
198 password
: str = None,
200 raise MethodNotImplemented()
202 async def repo_list(self
):
203 raise MethodNotImplemented()
205 async def repo_remove(
209 raise MethodNotImplemented()
211 async def synchronize_repos(self
, cluster_uuid
: str, name
: str):
213 Returns None as currently add_repo is not implemented
223 uninstall_sw
: bool = False,
228 Resets the Kubernetes cluster by removing the model that represents it.
230 :param cluster_uuid str: The UUID of the cluster to reset
231 :param force: Force reset
232 :param uninstall_sw: Boolean to uninstall sw
233 :param: kwargs: Additional parameters
236 :return: Returns True if successful or raises an exception.
240 self
.log
.debug("[reset] Removing k8s cloud")
241 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
243 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
245 cloud_creds
= await libjuju
.get_cloud_credentials(cloud
)
247 await libjuju
.remove_cloud(cluster_uuid
)
249 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
251 kubectl
= self
._get
_kubectl
(credentials
)
254 kubectl
.delete_cluster_role_binding
,
255 kubectl
.delete_service_account
,
256 kubectl
.delete_cluster_role
,
259 credential_attrs
= cloud_creds
[0].result
["attrs"]
260 if RBAC_LABEL_KEY_NAME
in credential_attrs
:
261 rbac_id
= credential_attrs
[RBAC_LABEL_KEY_NAME
]
262 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
263 for delete_func
in delete_functions
:
265 delete_func(metadata_name
)
266 except Exception as e
:
267 self
.log
.warning("Cannot remove resource in K8s {}".format(e
))
269 except Exception as e
:
270 self
.log
.debug("Caught exception during reset: {}".format(e
))
282 timeout
: float = 1800,
284 db_dict
: dict = None,
285 kdu_name
: str = None,
286 namespace
: str = None,
291 :param cluster_uuid str: The UUID of the cluster to install to
292 :param kdu_model str: The name or path of a bundle to install
293 :param kdu_instance: Kdu instance name
294 :param atomic bool: If set, waits until the model is active and resets
295 the cluster on failure.
296 :param timeout int: The time, in seconds, to wait for the install
298 :param params dict: Key-value pairs of instantiation parameters
299 :param kdu_name: Name of the KDU instance to be installed
300 :param namespace: K8s namespace to use for the KDU instance
301 :param kwargs: Additional parameters
304 :return: If successful, returns ?
306 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
310 raise K8sException("db_dict must be set")
312 raise K8sException("bundle must be set")
314 if bundle
.startswith("cs:"):
315 # For Juju Bundles provided by the Charm Store
317 elif bundle
.startswith("ch:"):
318 # For Juju Bundles provided by the Charm Hub (this only works for juju version >= 2.9)
320 elif bundle
.startswith("http"):
324 new_workdir
= kdu_model
.strip(kdu_model
.split("/")[-1])
325 os
.chdir(new_workdir
)
326 bundle
= "local:{}".format(kdu_model
)
328 self
.log
.debug("Checking for model named {}".format(kdu_instance
))
330 # Create the new model
331 self
.log
.debug("Adding model: {}".format(kdu_instance
))
332 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
333 await libjuju
.add_model(kdu_instance
, cloud
)
336 # TODO: Instantiation parameters
339 "Juju bundle that models the KDU, in any of the following ways:
340 - <juju-repo>/<juju-bundle>
341 - <juju-bundle folder under k8s_models folder in the package>
342 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
344 - <URL_where_to_fetch_juju_bundle>
347 previous_workdir
= os
.getcwd()
348 except FileNotFoundError
:
349 previous_workdir
= "/app/storage"
351 self
.log
.debug("[install] deploying {}".format(bundle
))
352 await libjuju
.deploy(
353 bundle
, model_name
=kdu_instance
, wait
=atomic
, timeout
=timeout
355 os
.chdir(previous_workdir
)
356 if self
.on_update_db
:
357 await self
.on_update_db(
360 filter=db_dict
["filter"],
361 vca_id
=kwargs
.get("vca_id"),
370 total_timeout
: float = 1800,
373 """Scale an application in a model
375 :param: kdu_instance str: KDU instance name
376 :param: scale int: Scale to which to set the application
377 :param: resource_name str: The application name in the Juju Bundle
378 :param: timeout float: The time, in seconds, to wait for the install
380 :param kwargs: Additional parameters
383 :return: If successful, returns True
387 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
388 await libjuju
.scale_application(
389 model_name
=kdu_instance
,
390 application_name
=resource_name
,
392 total_timeout
=total_timeout
,
394 except Exception as e
:
395 error_msg
= "Error scaling application {} in kdu instance {}: {}".format(
396 resource_name
, kdu_instance
, e
398 self
.log
.error(error_msg
)
399 raise K8sException(message
=error_msg
)
402 async def get_scale_count(
408 """Get an application scale count
410 :param: resource_name str: The application name in the Juju Bundle
411 :param: kdu_instance str: KDU instance name
412 :param kwargs: Additional parameters
414 :return: Return application instance count
418 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
419 status
= await libjuju
.get_model_status(kdu_instance
)
420 return len(status
.applications
[resource_name
].units
)
421 except Exception as e
:
422 error_msg
= "Error getting scale count from application {} in kdu instance {}: {}".format(
423 resource_name
, kdu_instance
, e
425 self
.log
.error(error_msg
)
426 raise K8sException(message
=error_msg
)
428 async def instances_list(self
, cluster_uuid
: str) -> list:
430 returns a list of deployed releases in a cluster
432 :param cluster_uuid: the cluster
441 kdu_model
: str = None,
446 :param cluster_uuid str: The UUID of the cluster to upgrade
447 :param kdu_instance str: The unique name of the KDU instance
448 :param kdu_model str: The name or path of the bundle to upgrade to
449 :param params dict: Key-value pairs of instantiation parameters
451 :return: If successful, reference to the new revision number of the
455 # TODO: Loop through the bundle and upgrade each charm individually
458 The API doesn't have a concept of bundle upgrades, because there are
459 many possible changes: charm revision, disk, number of units, etc.
461 As such, we are only supporting a limited subset of upgrades. We'll
462 upgrade the charm revision but leave storage and scale untouched.
464 Scale changes should happen through OSM constructs, and changes to
465 storage would require a redeployment of the service, at least in this
468 raise MethodNotImplemented()
480 :param cluster_uuid str: The UUID of the cluster to rollback
481 :param kdu_instance str: The unique name of the KDU instance
482 :param revision int: The revision to revert to. If omitted, rolls back
483 the previous upgrade.
485 :return: If successful, returns the revision of active KDU instance,
486 or raises an exception
488 raise MethodNotImplemented()
498 """Uninstall a KDU instance
500 :param cluster_uuid str: The UUID of the cluster
501 :param kdu_instance str: The unique name of the KDU instance
502 :param kwargs: Additional parameters
505 :return: Returns True if successful, or raises an exception
508 self
.log
.debug("[uninstall] Destroying model")
509 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
511 await libjuju
.destroy_model(kdu_instance
, total_timeout
=3600)
513 # self.log.debug("[uninstall] Model destroyed and disconnecting")
514 # await controller.disconnect()
517 # TODO: Remove these commented lines
518 # if not self.authenticated:
519 # self.log.debug("[uninstall] Connecting to controller")
520 # await self.login(cluster_uuid)
522 async def exec_primitive(
524 cluster_uuid
: str = None,
525 kdu_instance
: str = None,
526 primitive_name
: str = None,
527 timeout
: float = 300,
529 db_dict
: dict = None,
532 """Exec primitive (Juju action)
534 :param cluster_uuid str: The UUID of the cluster
535 :param kdu_instance str: The unique name of the KDU instance
536 :param primitive_name: Name of action that will be executed
537 :param timeout: Timeout for action execution
538 :param params: Dictionary of all the parameters needed for the action
539 :param db_dict: Dictionary for any additional data
540 :param kwargs: Additional parameters
543 :return: Returns the output of the action
545 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
547 if not params
or "application-name" not in params
:
549 "Missing application-name argument, \
550 argument needed for K8s actions"
554 "[exec_primitive] Getting model "
555 "kdu_instance: {}".format(kdu_instance
)
557 application_name
= params
["application-name"]
558 actions
= await libjuju
.get_actions(application_name
, kdu_instance
)
559 if primitive_name
not in actions
:
560 raise K8sException("Primitive {} not found".format(primitive_name
))
561 output
, status
= await libjuju
.execute_action(
562 application_name
, kdu_instance
, primitive_name
, **params
565 if status
!= "completed":
567 "status is not completed: {} output: {}".format(status
, output
)
569 if self
.on_update_db
:
570 await self
.on_update_db(
571 cluster_uuid
, kdu_instance
, filter=db_dict
["filter"]
576 except Exception as e
:
577 error_msg
= "Error executing primitive {}: {}".format(primitive_name
, e
)
578 self
.log
.error(error_msg
)
579 raise K8sException(message
=error_msg
)
583 async def inspect_kdu(
589 Inspects a bundle and returns a dictionary of config parameters and
590 their default values.
592 :param kdu_model str: The name or path of the bundle to inspect.
594 :return: If successful, returns a dictionary of available parameters
595 and their default values.
599 if not os
.path
.exists(kdu_model
):
600 raise K8sException("file {} not found".format(kdu_model
))
602 with
open(kdu_model
, "r") as f
:
603 bundle
= yaml
.safe_load(f
.read())
607 'description': 'Test bundle',
608 'bundle': 'kubernetes',
611 'charm': 'cs:~charmed-osm/mariadb-k8s-20',
614 'password': 'manopw',
615 'root_password': 'osm4u',
618 'series': 'kubernetes'
623 # TODO: This should be returned in an agreed-upon format
624 kdu
= bundle
["applications"]
634 If available, returns the README of the bundle.
636 :param kdu_model str: The name or path of a bundle
638 :return: If found, returns the contents of the README.
642 files
= ["README", "README.txt", "README.md"]
643 path
= os
.path
.dirname(kdu_model
)
644 for file in os
.listdir(path
):
646 with
open(file, "r") as f
:
652 async def status_kdu(
656 complete_status
: bool = False,
657 yaml_format
: bool = False,
660 """Get the status of the KDU
662 Get the current status of the KDU instance.
664 :param cluster_uuid str: The UUID of the cluster
665 :param kdu_instance str: The unique id of the KDU instance
666 :param complete_status: To get the complete_status of the KDU
667 :param yaml_format: To get the status in proper format for NSR record
668 :param: kwargs: Additional parameters
671 :return: Returns a dictionary containing namespace, state, resources,
672 and deployment_time and returns complete_status if complete_status is True
674 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
677 model_status
= await libjuju
.get_model_status(kdu_instance
)
679 if not complete_status
:
680 for name
in model_status
.applications
:
681 application
= model_status
.applications
[name
]
682 status
[name
] = {"status": application
["status"]["status"]}
685 return obj_to_yaml(model_status
)
687 return obj_to_dict(model_status
)
691 async def add_relation(
693 provider
: RelationEndpoint
,
694 requirer
: RelationEndpoint
,
697 Add relation between two charmed endpoints
699 :param: provider: Provider relation endpoint
700 :param: requirer: Requirer relation endpoint
702 self
.log
.debug(f
"adding new relation between {provider} and {requirer}")
703 cross_model_relation
= (
704 provider
.model_name
!= requirer
.model_name
705 or requirer
.vca_id
!= requirer
.vca_id
708 if cross_model_relation
:
709 # Cross-model relation
710 provider_libjuju
= await self
._get
_libjuju
(provider
.vca_id
)
711 requirer_libjuju
= await self
._get
_libjuju
(requirer
.vca_id
)
712 offer
= await provider_libjuju
.offer(provider
)
714 saas_name
= await requirer_libjuju
.consume(
715 requirer
.model_name
, offer
, provider_libjuju
717 await requirer_libjuju
.add_relation(
724 vca_id
= provider
.vca_id
725 model
= provider
.model_name
726 libjuju
= await self
._get
_libjuju
(vca_id
)
727 # add juju relations between two applications
728 await libjuju
.add_relation(
730 endpoint_1
=provider
.endpoint
,
731 endpoint_2
=requirer
.endpoint
,
733 except Exception as e
:
734 message
= f
"Error adding relation between {provider} and {requirer}: {e}"
735 self
.log
.error(message
)
736 raise Exception(message
=message
)
738 async def update_vca_status(self
, vcastatus
: dict, kdu_instance
: str, **kwargs
):
740 Add all configs, actions, executed actions of all applications in a model to vcastatus dict
742 :param vcastatus dict: dict containing vcastatus
743 :param kdu_instance str: The unique id of the KDU instance
744 :param: kwargs: Additional parameters
749 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
751 for model_name
in vcastatus
:
752 # Adding executed actions
753 vcastatus
[model_name
][
755 ] = await libjuju
.get_executed_actions(kdu_instance
)
757 for application
in vcastatus
[model_name
]["applications"]:
758 # Adding application actions
759 vcastatus
[model_name
]["applications"][application
][
761 ] = await libjuju
.get_actions(application
, kdu_instance
)
762 # Adding application configs
763 vcastatus
[model_name
]["applications"][application
][
765 ] = await libjuju
.get_application_configs(kdu_instance
, application
)
767 except Exception as e
:
768 self
.log
.debug("Error in updating vca status: {}".format(str(e
)))
770 async def get_services(
771 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
773 """Return a list of services of a kdu_instance"""
775 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
776 kubectl
= self
._get
_kubectl
(credentials
)
777 return kubectl
.get_services(
778 field_selector
="metadata.namespace={}".format(kdu_instance
)
781 async def get_service(
782 self
, cluster_uuid
: str, service_name
: str, namespace
: str
784 """Return data for a specific service inside a namespace"""
786 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
787 kubectl
= self
._get
_kubectl
(credentials
)
788 return kubectl
.get_services(
789 field_selector
="metadata.name={},metadata.namespace={}".format(
790 service_name
, namespace
794 def get_credentials(self
, cluster_uuid
: str) -> str:
796 Get Cluster Kubeconfig
798 k8scluster
= self
.db
.get_one(
799 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
802 self
.db
.encrypt_decrypt_fields(
803 k8scluster
.get("credentials"),
805 ["password", "secret"],
806 schema_version
=k8scluster
["schema_version"],
807 salt
=k8scluster
["_id"],
810 return yaml
.safe_dump(k8scluster
.get("credentials"))
812 def _get_credential_name(self
, cluster_uuid
: str) -> str:
814 Get credential name for a k8s cloud
816 We cannot use the cluster_uuid for the credential name directly,
817 because it cannot start with a number, it must start with a letter.
818 Therefore, the k8s cloud credential name will be "cred-" followed
821 :param: cluster_uuid: Cluster UUID of the kubernetes cloud (=cloud_name)
823 :return: Name to use for the credential name.
825 return "cred-{}".format(cluster_uuid
)
831 """Get the namespace UUID
832 Gets the namespace's unique name
834 :param cluster_uuid str: The UUID of the cluster
835 :returns: The namespace UUID, or raises an exception
840 def generate_kdu_instance_name(**kwargs
):
841 db_dict
= kwargs
.get("db_dict")
842 kdu_name
= kwargs
.get("kdu_name", None)
844 kdu_instance
= "{}-{}".format(kdu_name
, db_dict
["filter"]["_id"])
846 kdu_instance
= db_dict
["filter"]["_id"]
849 async def _get_libjuju(self
, vca_id
: str = None) -> Libjuju
:
853 :param: vca_id: VCA ID
854 If None, get a libjuju object with a Connection to the default VCA
855 Else, geta libjuju object with a Connection to the specified VCA
858 while self
.loading_libjuju
.locked():
859 await asyncio
.sleep(0.1)
861 async with self
.loading_libjuju
:
862 vca_connection
= await get_connection(self
._store
)
863 self
.libjuju
= Libjuju(vca_connection
, loop
=self
.loop
, log
=self
.log
)
866 vca_connection
= await get_connection(self
._store
, vca_id
)
874 def _get_kubectl(self
, credentials
: str) -> Kubectl
:
878 :param: kubeconfig_credentials: Kubeconfig credentials
880 kubecfg
= tempfile
.NamedTemporaryFile()
881 with
open(kubecfg
.name
, "w") as kubecfg_file
:
882 kubecfg_file
.write(credentials
)
883 return Kubectl(config_file
=kubecfg
.name
)