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",
197 raise MethodNotImplemented()
199 async def repo_list(self
):
200 raise MethodNotImplemented()
202 async def repo_remove(
206 raise MethodNotImplemented()
208 async def synchronize_repos(self
, cluster_uuid
: str, name
: str):
210 Returns None as currently add_repo is not implemented
220 uninstall_sw
: bool = False,
225 Resets the Kubernetes cluster by removing the model that represents it.
227 :param cluster_uuid str: The UUID of the cluster to reset
228 :param force: Force reset
229 :param uninstall_sw: Boolean to uninstall sw
230 :param: kwargs: Additional parameters
233 :return: Returns True if successful or raises an exception.
237 self
.log
.debug("[reset] Removing k8s cloud")
238 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
240 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
242 cloud_creds
= await libjuju
.get_cloud_credentials(cloud
)
244 await libjuju
.remove_cloud(cluster_uuid
)
246 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
248 kubectl
= self
._get
_kubectl
(credentials
)
251 kubectl
.delete_cluster_role_binding
,
252 kubectl
.delete_service_account
,
253 kubectl
.delete_cluster_role
,
256 credential_attrs
= cloud_creds
[0].result
["attrs"]
257 if RBAC_LABEL_KEY_NAME
in credential_attrs
:
258 rbac_id
= credential_attrs
[RBAC_LABEL_KEY_NAME
]
259 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
260 for delete_func
in delete_functions
:
262 delete_func(metadata_name
)
263 except Exception as e
:
264 self
.log
.warning("Cannot remove resource in K8s {}".format(e
))
266 except Exception as e
:
267 self
.log
.debug("Caught exception during reset: {}".format(e
))
279 timeout
: float = 1800,
281 db_dict
: dict = None,
282 kdu_name
: str = None,
283 namespace
: str = None,
288 :param cluster_uuid str: The UUID of the cluster to install to
289 :param kdu_model str: The name or path of a bundle to install
290 :param kdu_instance: Kdu instance name
291 :param atomic bool: If set, waits until the model is active and resets
292 the cluster on failure.
293 :param timeout int: The time, in seconds, to wait for the install
295 :param params dict: Key-value pairs of instantiation parameters
296 :param kdu_name: Name of the KDU instance to be installed
297 :param namespace: K8s namespace to use for the KDU instance
298 :param kwargs: Additional parameters
301 :return: If successful, returns ?
303 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
307 raise K8sException("db_dict must be set")
309 raise K8sException("bundle must be set")
311 if bundle
.startswith("cs:"):
312 # For Juju Bundles provided by the Charm Store
314 elif bundle
.startswith("ch:"):
315 # For Juju Bundles provided by the Charm Hub (this only works for juju version >= 2.9)
317 elif bundle
.startswith("http"):
321 new_workdir
= kdu_model
.strip(kdu_model
.split("/")[-1])
322 os
.chdir(new_workdir
)
323 bundle
= "local:{}".format(kdu_model
)
325 self
.log
.debug("Checking for model named {}".format(kdu_instance
))
327 # Create the new model
328 self
.log
.debug("Adding model: {}".format(kdu_instance
))
329 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
330 await libjuju
.add_model(kdu_instance
, cloud
)
333 # TODO: Instantiation parameters
336 "Juju bundle that models the KDU, in any of the following ways:
337 - <juju-repo>/<juju-bundle>
338 - <juju-bundle folder under k8s_models folder in the package>
339 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
341 - <URL_where_to_fetch_juju_bundle>
344 previous_workdir
= os
.getcwd()
345 except FileNotFoundError
:
346 previous_workdir
= "/app/storage"
348 self
.log
.debug("[install] deploying {}".format(bundle
))
349 await libjuju
.deploy(
350 bundle
, model_name
=kdu_instance
, wait
=atomic
, timeout
=timeout
352 os
.chdir(previous_workdir
)
353 if self
.on_update_db
:
354 await self
.on_update_db(
357 filter=db_dict
["filter"],
358 vca_id
=kwargs
.get("vca_id"),
367 total_timeout
: float = 1800,
370 """Scale an application in a model
372 :param: kdu_instance str: KDU instance name
373 :param: scale int: Scale to which to set the application
374 :param: resource_name str: The application name in the Juju Bundle
375 :param: timeout float: The time, in seconds, to wait for the install
377 :param kwargs: Additional parameters
380 :return: If successful, returns True
384 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
385 await libjuju
.scale_application(
386 model_name
=kdu_instance
,
387 application_name
=resource_name
,
389 total_timeout
=total_timeout
,
391 except Exception as e
:
392 error_msg
= "Error scaling application {} in kdu instance {}: {}".format(
393 resource_name
, kdu_instance
, e
395 self
.log
.error(error_msg
)
396 raise K8sException(message
=error_msg
)
399 async def get_scale_count(
405 """Get an application scale count
407 :param: resource_name str: The application name in the Juju Bundle
408 :param: kdu_instance str: KDU instance name
409 :param kwargs: Additional parameters
411 :return: Return application instance count
415 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
416 status
= await libjuju
.get_model_status(kdu_instance
)
417 return len(status
.applications
[resource_name
].units
)
418 except Exception as e
:
419 error_msg
= "Error getting scale count from application {} in kdu instance {}: {}".format(
420 resource_name
, kdu_instance
, e
422 self
.log
.error(error_msg
)
423 raise K8sException(message
=error_msg
)
425 async def instances_list(self
, cluster_uuid
: str) -> list:
427 returns a list of deployed releases in a cluster
429 :param cluster_uuid: the cluster
438 kdu_model
: str = None,
443 :param cluster_uuid str: The UUID of the cluster to upgrade
444 :param kdu_instance str: The unique name of the KDU instance
445 :param kdu_model str: The name or path of the bundle to upgrade to
446 :param params dict: Key-value pairs of instantiation parameters
448 :return: If successful, reference to the new revision number of the
452 # TODO: Loop through the bundle and upgrade each charm individually
455 The API doesn't have a concept of bundle upgrades, because there are
456 many possible changes: charm revision, disk, number of units, etc.
458 As such, we are only supporting a limited subset of upgrades. We'll
459 upgrade the charm revision but leave storage and scale untouched.
461 Scale changes should happen through OSM constructs, and changes to
462 storage would require a redeployment of the service, at least in this
465 raise MethodNotImplemented()
477 :param cluster_uuid str: The UUID of the cluster to rollback
478 :param kdu_instance str: The unique name of the KDU instance
479 :param revision int: The revision to revert to. If omitted, rolls back
480 the previous upgrade.
482 :return: If successful, returns the revision of active KDU instance,
483 or raises an exception
485 raise MethodNotImplemented()
495 """Uninstall a KDU instance
497 :param cluster_uuid str: The UUID of the cluster
498 :param kdu_instance str: The unique name of the KDU instance
499 :param kwargs: Additional parameters
502 :return: Returns True if successful, or raises an exception
505 self
.log
.debug("[uninstall] Destroying model")
506 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
508 await libjuju
.destroy_model(kdu_instance
, total_timeout
=3600)
510 # self.log.debug("[uninstall] Model destroyed and disconnecting")
511 # await controller.disconnect()
514 # TODO: Remove these commented lines
515 # if not self.authenticated:
516 # self.log.debug("[uninstall] Connecting to controller")
517 # await self.login(cluster_uuid)
519 async def exec_primitive(
521 cluster_uuid
: str = None,
522 kdu_instance
: str = None,
523 primitive_name
: str = None,
524 timeout
: float = 300,
526 db_dict
: dict = None,
529 """Exec primitive (Juju action)
531 :param cluster_uuid str: The UUID of the cluster
532 :param kdu_instance str: The unique name of the KDU instance
533 :param primitive_name: Name of action that will be executed
534 :param timeout: Timeout for action execution
535 :param params: Dictionary of all the parameters needed for the action
536 :param db_dict: Dictionary for any additional data
537 :param kwargs: Additional parameters
540 :return: Returns the output of the action
542 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
544 if not params
or "application-name" not in params
:
546 "Missing application-name argument, \
547 argument needed for K8s actions"
551 "[exec_primitive] Getting model "
552 "kdu_instance: {}".format(kdu_instance
)
554 application_name
= params
["application-name"]
555 actions
= await libjuju
.get_actions(application_name
, kdu_instance
)
556 if primitive_name
not in actions
:
557 raise K8sException("Primitive {} not found".format(primitive_name
))
558 output
, status
= await libjuju
.execute_action(
559 application_name
, kdu_instance
, primitive_name
, **params
562 if status
!= "completed":
564 "status is not completed: {} output: {}".format(status
, output
)
566 if self
.on_update_db
:
567 await self
.on_update_db(
568 cluster_uuid
, kdu_instance
, filter=db_dict
["filter"]
573 except Exception as e
:
574 error_msg
= "Error executing primitive {}: {}".format(primitive_name
, e
)
575 self
.log
.error(error_msg
)
576 raise K8sException(message
=error_msg
)
580 async def inspect_kdu(
586 Inspects a bundle and returns a dictionary of config parameters and
587 their default values.
589 :param kdu_model str: The name or path of the bundle to inspect.
591 :return: If successful, returns a dictionary of available parameters
592 and their default values.
596 if not os
.path
.exists(kdu_model
):
597 raise K8sException("file {} not found".format(kdu_model
))
599 with
open(kdu_model
, "r") as f
:
600 bundle
= yaml
.safe_load(f
.read())
604 'description': 'Test bundle',
605 'bundle': 'kubernetes',
608 'charm': 'cs:~charmed-osm/mariadb-k8s-20',
611 'password': 'manopw',
612 'root_password': 'osm4u',
615 'series': 'kubernetes'
620 # TODO: This should be returned in an agreed-upon format
621 kdu
= bundle
["applications"]
631 If available, returns the README of the bundle.
633 :param kdu_model str: The name or path of a bundle
635 :return: If found, returns the contents of the README.
639 files
= ["README", "README.txt", "README.md"]
640 path
= os
.path
.dirname(kdu_model
)
641 for file in os
.listdir(path
):
643 with
open(file, "r") as f
:
649 async def status_kdu(
653 complete_status
: bool = False,
654 yaml_format
: bool = False,
657 """Get the status of the KDU
659 Get the current status of the KDU instance.
661 :param cluster_uuid str: The UUID of the cluster
662 :param kdu_instance str: The unique id of the KDU instance
663 :param complete_status: To get the complete_status of the KDU
664 :param yaml_format: To get the status in proper format for NSR record
665 :param: kwargs: Additional parameters
668 :return: Returns a dictionary containing namespace, state, resources,
669 and deployment_time and returns complete_status if complete_status is True
671 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
674 model_status
= await libjuju
.get_model_status(kdu_instance
)
676 if not complete_status
:
677 for name
in model_status
.applications
:
678 application
= model_status
.applications
[name
]
679 status
[name
] = {"status": application
["status"]["status"]}
682 return obj_to_yaml(model_status
)
684 return obj_to_dict(model_status
)
688 async def add_relation(
690 provider
: RelationEndpoint
,
691 requirer
: RelationEndpoint
,
694 Add relation between two charmed endpoints
696 :param: provider: Provider relation endpoint
697 :param: requirer: Requirer relation endpoint
699 self
.log
.debug(f
"adding new relation between {provider} and {requirer}")
700 cross_model_relation
= (
701 provider
.model_name
!= requirer
.model_name
702 or requirer
.vca_id
!= requirer
.vca_id
705 if cross_model_relation
:
706 # Cross-model relation
707 provider_libjuju
= await self
._get
_libjuju
(provider
.vca_id
)
708 requirer_libjuju
= await self
._get
_libjuju
(requirer
.vca_id
)
709 offer
= await provider_libjuju
.offer(provider
)
711 saas_name
= await requirer_libjuju
.consume(
712 requirer
.model_name
, offer
, provider_libjuju
714 await requirer_libjuju
.add_relation(
721 vca_id
= provider
.vca_id
722 model
= provider
.model_name
723 libjuju
= await self
._get
_libjuju
(vca_id
)
724 # add juju relations between two applications
725 await libjuju
.add_relation(
727 endpoint_1
=provider
.endpoint
,
728 endpoint_2
=requirer
.endpoint
,
730 except Exception as e
:
731 message
= f
"Error adding relation between {provider} and {requirer}: {e}"
732 self
.log
.error(message
)
733 raise Exception(message
=message
)
735 async def update_vca_status(self
, vcastatus
: dict, kdu_instance
: str, **kwargs
):
737 Add all configs, actions, executed actions of all applications in a model to vcastatus dict
739 :param vcastatus dict: dict containing vcastatus
740 :param kdu_instance str: The unique id of the KDU instance
741 :param: kwargs: Additional parameters
746 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
748 for model_name
in vcastatus
:
749 # Adding executed actions
750 vcastatus
[model_name
][
752 ] = await libjuju
.get_executed_actions(kdu_instance
)
754 for application
in vcastatus
[model_name
]["applications"]:
755 # Adding application actions
756 vcastatus
[model_name
]["applications"][application
][
758 ] = await libjuju
.get_actions(application
, kdu_instance
)
759 # Adding application configs
760 vcastatus
[model_name
]["applications"][application
][
762 ] = await libjuju
.get_application_configs(kdu_instance
, application
)
764 except Exception as e
:
765 self
.log
.debug("Error in updating vca status: {}".format(str(e
)))
767 async def get_services(
768 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
770 """Return a list of services of a kdu_instance"""
772 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
773 kubectl
= self
._get
_kubectl
(credentials
)
774 return kubectl
.get_services(
775 field_selector
="metadata.namespace={}".format(kdu_instance
)
778 async def get_service(
779 self
, cluster_uuid
: str, service_name
: str, namespace
: str
781 """Return data for a specific service inside a namespace"""
783 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
784 kubectl
= self
._get
_kubectl
(credentials
)
785 return kubectl
.get_services(
786 field_selector
="metadata.name={},metadata.namespace={}".format(
787 service_name
, namespace
791 def get_credentials(self
, cluster_uuid
: str) -> str:
793 Get Cluster Kubeconfig
795 k8scluster
= self
.db
.get_one(
796 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
799 self
.db
.encrypt_decrypt_fields(
800 k8scluster
.get("credentials"),
802 ["password", "secret"],
803 schema_version
=k8scluster
["schema_version"],
804 salt
=k8scluster
["_id"],
807 return yaml
.safe_dump(k8scluster
.get("credentials"))
809 def _get_credential_name(self
, cluster_uuid
: str) -> str:
811 Get credential name for a k8s cloud
813 We cannot use the cluster_uuid for the credential name directly,
814 because it cannot start with a number, it must start with a letter.
815 Therefore, the k8s cloud credential name will be "cred-" followed
818 :param: cluster_uuid: Cluster UUID of the kubernetes cloud (=cloud_name)
820 :return: Name to use for the credential name.
822 return "cred-{}".format(cluster_uuid
)
828 """Get the namespace UUID
829 Gets the namespace's unique name
831 :param cluster_uuid str: The UUID of the cluster
832 :returns: The namespace UUID, or raises an exception
837 def generate_kdu_instance_name(**kwargs
):
838 db_dict
= kwargs
.get("db_dict")
839 kdu_name
= kwargs
.get("kdu_name", None)
841 kdu_instance
= "{}-{}".format(kdu_name
, db_dict
["filter"]["_id"])
843 kdu_instance
= db_dict
["filter"]["_id"]
846 async def _get_libjuju(self
, vca_id
: str = None) -> Libjuju
:
850 :param: vca_id: VCA ID
851 If None, get a libjuju object with a Connection to the default VCA
852 Else, geta libjuju object with a Connection to the specified VCA
855 while self
.loading_libjuju
.locked():
856 await asyncio
.sleep(0.1)
858 async with self
.loading_libjuju
:
859 vca_connection
= await get_connection(self
._store
)
860 self
.libjuju
= Libjuju(vca_connection
, loop
=self
.loop
, log
=self
.log
)
863 vca_connection
= await get_connection(self
._store
, vca_id
)
871 def _get_kubectl(self
, credentials
: str) -> Kubectl
:
875 :param: kubeconfig_credentials: Kubeconfig credentials
877 kubecfg
= tempfile
.NamedTemporaryFile()
878 with
open(kubecfg
.name
, "w") as kubecfg_file
:
879 kubecfg_file
.write(credentials
)
880 return Kubectl(config_file
=kubecfg
.name
)