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 kubectl
.create_cluster_role(
131 "delete": kubectl
.delete_cluster_role
,
132 "args": (metadata_name
,),
136 kubectl
.create_service_account(
142 "delete": kubectl
.delete_service_account
,
143 "args": (metadata_name
,),
147 kubectl
.create_cluster_role_binding(
153 "delete": kubectl
.delete_service_account
,
154 "args": (metadata_name
,),
157 token
, client_cert_data
= await kubectl
.get_secret_data(
161 default_storage_class
= kubectl
.get_default_storage_class()
162 await libjuju
.add_k8s(
166 client_cert_data
=client_cert_data
,
167 configuration
=kubectl
.configuration
,
168 storage_class
=default_storage_class
,
169 credential_name
=self
._get
_credential
_name
(cluster_uuid
),
171 return cluster_uuid
, True
172 except Exception as e
:
173 self
.log
.error("Error initializing k8scluster: {}".format(e
))
174 if len(cleanup_data
) > 0:
175 self
.log
.debug("Cleaning up created resources in k8s cluster...")
176 for item
in cleanup_data
:
177 delete_function
= item
["delete"]
178 delete_args
= item
["args"]
179 delete_function(*delete_args
)
180 self
.log
.debug("Cleanup finished")
183 """Repo Management"""
189 _type
: str = "charm",
191 raise MethodNotImplemented()
193 async def repo_list(self
):
194 raise MethodNotImplemented()
196 async def repo_remove(
200 raise MethodNotImplemented()
202 async def synchronize_repos(self
, cluster_uuid
: str, name
: str):
204 Returns None as currently add_repo is not implemented
214 uninstall_sw
: bool = False,
219 Resets the Kubernetes cluster by removing the model that represents it.
221 :param cluster_uuid str: The UUID of the cluster to reset
222 :param force: Force reset
223 :param uninstall_sw: Boolean to uninstall sw
224 :param: kwargs: Additional parameters
227 :return: Returns True if successful or raises an exception.
231 self
.log
.debug("[reset] Removing k8s cloud")
232 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
234 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
236 cloud_creds
= await libjuju
.get_cloud_credentials(cloud
)
238 await libjuju
.remove_cloud(cluster_uuid
)
240 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
242 kubectl
= self
._get
_kubectl
(credentials
)
245 kubectl
.delete_cluster_role_binding
,
246 kubectl
.delete_service_account
,
247 kubectl
.delete_cluster_role
,
250 credential_attrs
= cloud_creds
[0].result
["attrs"]
251 if RBAC_LABEL_KEY_NAME
in credential_attrs
:
252 rbac_id
= credential_attrs
[RBAC_LABEL_KEY_NAME
]
253 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
254 for delete_func
in delete_functions
:
256 delete_func(metadata_name
)
257 except Exception as e
:
258 self
.log
.warning("Cannot remove resource in K8s {}".format(e
))
260 except Exception as e
:
261 self
.log
.debug("Caught exception during reset: {}".format(e
))
273 timeout
: float = 1800,
275 db_dict
: dict = None,
276 kdu_name
: str = None,
277 namespace
: str = None,
282 :param cluster_uuid str: The UUID of the cluster to install to
283 :param kdu_model str: The name or path of a bundle to install
284 :param kdu_instance: Kdu instance name
285 :param atomic bool: If set, waits until the model is active and resets
286 the cluster on failure.
287 :param timeout int: The time, in seconds, to wait for the install
289 :param params dict: Key-value pairs of instantiation parameters
290 :param kdu_name: Name of the KDU instance to be installed
291 :param namespace: K8s namespace to use for the KDU instance
292 :param kwargs: Additional parameters
295 :return: If successful, returns ?
297 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
301 raise K8sException("db_dict must be set")
303 raise K8sException("bundle must be set")
305 if bundle
.startswith("cs:"):
307 elif bundle
.startswith("http"):
311 new_workdir
= kdu_model
.strip(kdu_model
.split("/")[-1])
312 os
.chdir(new_workdir
)
313 bundle
= "local:{}".format(kdu_model
)
315 self
.log
.debug("Checking for model named {}".format(kdu_instance
))
317 # Create the new model
318 self
.log
.debug("Adding model: {}".format(kdu_instance
))
319 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
320 await libjuju
.add_model(kdu_instance
, cloud
)
323 # TODO: Instantiation parameters
326 "Juju bundle that models the KDU, in any of the following ways:
327 - <juju-repo>/<juju-bundle>
328 - <juju-bundle folder under k8s_models folder in the package>
329 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
331 - <URL_where_to_fetch_juju_bundle>
334 previous_workdir
= os
.getcwd()
335 except FileNotFoundError
:
336 previous_workdir
= "/app/storage"
338 self
.log
.debug("[install] deploying {}".format(bundle
))
339 await libjuju
.deploy(
340 bundle
, model_name
=kdu_instance
, wait
=atomic
, timeout
=timeout
342 os
.chdir(previous_workdir
)
343 if self
.on_update_db
:
344 await self
.on_update_db(
347 filter=db_dict
["filter"],
348 vca_id
=kwargs
.get("vca_id"),
357 total_timeout
: float = 1800,
360 """Scale an application in a model
362 :param: kdu_instance str: KDU instance name
363 :param: scale int: Scale to which to set the application
364 :param: resource_name str: The application name in the Juju Bundle
365 :param: timeout float: The time, in seconds, to wait for the install
367 :param kwargs: Additional parameters
370 :return: If successful, returns True
374 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
375 await libjuju
.scale_application(
376 model_name
=kdu_instance
,
377 application_name
=resource_name
,
379 total_timeout
=total_timeout
,
381 except Exception as e
:
382 error_msg
= "Error scaling application {} in kdu instance {}: {}".format(
383 resource_name
, kdu_instance
, e
385 self
.log
.error(error_msg
)
386 raise K8sException(message
=error_msg
)
389 async def get_scale_count(
395 """Get an application scale count
397 :param: resource_name str: The application name in the Juju Bundle
398 :param: kdu_instance str: KDU instance name
399 :param kwargs: Additional parameters
401 :return: Return application instance count
405 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
406 status
= await libjuju
.get_model_status(kdu_instance
)
407 return len(status
.applications
[resource_name
].units
)
408 except Exception as e
:
409 error_msg
= "Error getting scale count from application {} in kdu instance {}: {}".format(
410 resource_name
, kdu_instance
, e
412 self
.log
.error(error_msg
)
413 raise K8sException(message
=error_msg
)
415 async def instances_list(self
, cluster_uuid
: str) -> list:
417 returns a list of deployed releases in a cluster
419 :param cluster_uuid: the cluster
428 kdu_model
: str = None,
433 :param cluster_uuid str: The UUID of the cluster to upgrade
434 :param kdu_instance str: The unique name of the KDU instance
435 :param kdu_model str: The name or path of the bundle to upgrade to
436 :param params dict: Key-value pairs of instantiation parameters
438 :return: If successful, reference to the new revision number of the
442 # TODO: Loop through the bundle and upgrade each charm individually
445 The API doesn't have a concept of bundle upgrades, because there are
446 many possible changes: charm revision, disk, number of units, etc.
448 As such, we are only supporting a limited subset of upgrades. We'll
449 upgrade the charm revision but leave storage and scale untouched.
451 Scale changes should happen through OSM constructs, and changes to
452 storage would require a redeployment of the service, at least in this
455 raise MethodNotImplemented()
467 :param cluster_uuid str: The UUID of the cluster to rollback
468 :param kdu_instance str: The unique name of the KDU instance
469 :param revision int: The revision to revert to. If omitted, rolls back
470 the previous upgrade.
472 :return: If successful, returns the revision of active KDU instance,
473 or raises an exception
475 raise MethodNotImplemented()
485 """Uninstall a KDU instance
487 :param cluster_uuid str: The UUID of the cluster
488 :param kdu_instance str: The unique name of the KDU instance
489 :param kwargs: Additional parameters
492 :return: Returns True if successful, or raises an exception
495 self
.log
.debug("[uninstall] Destroying model")
496 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
498 await libjuju
.destroy_model(kdu_instance
, total_timeout
=3600)
500 # self.log.debug("[uninstall] Model destroyed and disconnecting")
501 # await controller.disconnect()
504 # TODO: Remove these commented lines
505 # if not self.authenticated:
506 # self.log.debug("[uninstall] Connecting to controller")
507 # await self.login(cluster_uuid)
509 async def exec_primitive(
511 cluster_uuid
: str = None,
512 kdu_instance
: str = None,
513 primitive_name
: str = None,
514 timeout
: float = 300,
516 db_dict
: dict = None,
519 """Exec primitive (Juju action)
521 :param cluster_uuid str: The UUID of the cluster
522 :param kdu_instance str: The unique name of the KDU instance
523 :param primitive_name: Name of action that will be executed
524 :param timeout: Timeout for action execution
525 :param params: Dictionary of all the parameters needed for the action
526 :param db_dict: Dictionary for any additional data
527 :param kwargs: Additional parameters
530 :return: Returns the output of the action
532 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
534 if not params
or "application-name" not in params
:
536 "Missing application-name argument, \
537 argument needed for K8s actions"
541 "[exec_primitive] Getting model "
542 "kdu_instance: {}".format(kdu_instance
)
544 application_name
= params
["application-name"]
545 actions
= await libjuju
.get_actions(application_name
, kdu_instance
)
546 if primitive_name
not in actions
:
547 raise K8sException("Primitive {} not found".format(primitive_name
))
548 output
, status
= await libjuju
.execute_action(
549 application_name
, kdu_instance
, primitive_name
, **params
552 if status
!= "completed":
554 "status is not completed: {} output: {}".format(status
, output
)
556 if self
.on_update_db
:
557 await self
.on_update_db(
558 cluster_uuid
, kdu_instance
, filter=db_dict
["filter"]
563 except Exception as e
:
564 error_msg
= "Error executing primitive {}: {}".format(primitive_name
, e
)
565 self
.log
.error(error_msg
)
566 raise K8sException(message
=error_msg
)
570 async def inspect_kdu(
576 Inspects a bundle and returns a dictionary of config parameters and
577 their default values.
579 :param kdu_model str: The name or path of the bundle to inspect.
581 :return: If successful, returns a dictionary of available parameters
582 and their default values.
586 if not os
.path
.exists(kdu_model
):
587 raise K8sException("file {} not found".format(kdu_model
))
589 with
open(kdu_model
, "r") as f
:
590 bundle
= yaml
.safe_load(f
.read())
594 'description': 'Test bundle',
595 'bundle': 'kubernetes',
598 'charm': 'cs:~charmed-osm/mariadb-k8s-20',
601 'password': 'manopw',
602 'root_password': 'osm4u',
605 'series': 'kubernetes'
610 # TODO: This should be returned in an agreed-upon format
611 kdu
= bundle
["applications"]
621 If available, returns the README of the bundle.
623 :param kdu_model str: The name or path of a bundle
625 :return: If found, returns the contents of the README.
629 files
= ["README", "README.txt", "README.md"]
630 path
= os
.path
.dirname(kdu_model
)
631 for file in os
.listdir(path
):
633 with
open(file, "r") as f
:
639 async def status_kdu(
643 complete_status
: bool = False,
644 yaml_format
: bool = False,
647 """Get the status of the KDU
649 Get the current status of the KDU instance.
651 :param cluster_uuid str: The UUID of the cluster
652 :param kdu_instance str: The unique id of the KDU instance
653 :param complete_status: To get the complete_status of the KDU
654 :param yaml_format: To get the status in proper format for NSR record
655 :param: kwargs: Additional parameters
658 :return: Returns a dictionary containing namespace, state, resources,
659 and deployment_time and returns complete_status if complete_status is True
661 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
664 model_status
= await libjuju
.get_model_status(kdu_instance
)
666 if not complete_status
:
667 for name
in model_status
.applications
:
668 application
= model_status
.applications
[name
]
669 status
[name
] = {"status": application
["status"]["status"]}
672 return obj_to_yaml(model_status
)
674 return obj_to_dict(model_status
)
678 async def add_relation(
680 provider
: RelationEndpoint
,
681 requirer
: RelationEndpoint
,
684 Add relation between two charmed endpoints
686 :param: provider: Provider relation endpoint
687 :param: requirer: Requirer relation endpoint
689 self
.log
.debug(f
"adding new relation between {provider} and {requirer}")
690 cross_model_relation
= (
691 provider
.model_name
!= requirer
.model_name
692 or requirer
.vca_id
!= requirer
.vca_id
695 if cross_model_relation
:
696 # Cross-model relation
697 provider_libjuju
= await self
._get
_libjuju
(provider
.vca_id
)
698 requirer_libjuju
= await self
._get
_libjuju
(requirer
.vca_id
)
699 offer
= await provider_libjuju
.offer(provider
)
701 saas_name
= await requirer_libjuju
.consume(
702 requirer
.model_name
, offer
, provider_libjuju
704 await requirer_libjuju
.add_relation(
711 vca_id
= provider
.vca_id
712 model
= provider
.model_name
713 libjuju
= await self
._get
_libjuju
(vca_id
)
714 # add juju relations between two applications
715 await libjuju
.add_relation(
717 endpoint_1
=provider
.endpoint
,
718 endpoint_2
=requirer
.endpoint
,
720 except Exception as e
:
721 message
= f
"Error adding relation between {provider} and {requirer}: {e}"
722 self
.log
.error(message
)
723 raise Exception(message
=message
)
725 async def update_vca_status(self
, vcastatus
: dict, kdu_instance
: str, **kwargs
):
727 Add all configs, actions, executed actions of all applications in a model to vcastatus dict
729 :param vcastatus dict: dict containing vcastatus
730 :param kdu_instance str: The unique id of the KDU instance
731 :param: kwargs: Additional parameters
736 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
738 for model_name
in vcastatus
:
739 # Adding executed actions
740 vcastatus
[model_name
][
742 ] = await libjuju
.get_executed_actions(kdu_instance
)
744 for application
in vcastatus
[model_name
]["applications"]:
745 # Adding application actions
746 vcastatus
[model_name
]["applications"][application
][
748 ] = await libjuju
.get_actions(application
, kdu_instance
)
749 # Adding application configs
750 vcastatus
[model_name
]["applications"][application
][
752 ] = await libjuju
.get_application_configs(kdu_instance
, application
)
754 except Exception as e
:
755 self
.log
.debug("Error in updating vca status: {}".format(str(e
)))
757 async def get_services(
758 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
760 """Return a list of services of a kdu_instance"""
762 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
763 kubectl
= self
._get
_kubectl
(credentials
)
764 return kubectl
.get_services(
765 field_selector
="metadata.namespace={}".format(kdu_instance
)
768 async def get_service(
769 self
, cluster_uuid
: str, service_name
: str, namespace
: str
771 """Return data for a specific service inside a namespace"""
773 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
774 kubectl
= self
._get
_kubectl
(credentials
)
775 return kubectl
.get_services(
776 field_selector
="metadata.name={},metadata.namespace={}".format(
777 service_name
, namespace
781 def get_credentials(self
, cluster_uuid
: str) -> str:
783 Get Cluster Kubeconfig
785 k8scluster
= self
.db
.get_one(
786 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
789 self
.db
.encrypt_decrypt_fields(
790 k8scluster
.get("credentials"),
792 ["password", "secret"],
793 schema_version
=k8scluster
["schema_version"],
794 salt
=k8scluster
["_id"],
797 return yaml
.safe_dump(k8scluster
.get("credentials"))
799 def _get_credential_name(self
, cluster_uuid
: str) -> str:
801 Get credential name for a k8s cloud
803 We cannot use the cluster_uuid for the credential name directly,
804 because it cannot start with a number, it must start with a letter.
805 Therefore, the k8s cloud credential name will be "cred-" followed
808 :param: cluster_uuid: Cluster UUID of the kubernetes cloud (=cloud_name)
810 :return: Name to use for the credential name.
812 return "cred-{}".format(cluster_uuid
)
818 """Get the namespace UUID
819 Gets the namespace's unique name
821 :param cluster_uuid str: The UUID of the cluster
822 :returns: The namespace UUID, or raises an exception
827 def generate_kdu_instance_name(**kwargs
):
828 db_dict
= kwargs
.get("db_dict")
829 kdu_name
= kwargs
.get("kdu_name", None)
831 kdu_instance
= "{}-{}".format(kdu_name
, db_dict
["filter"]["_id"])
833 kdu_instance
= db_dict
["filter"]["_id"]
836 async def _get_libjuju(self
, vca_id
: str = None) -> Libjuju
:
840 :param: vca_id: VCA ID
841 If None, get a libjuju object with a Connection to the default VCA
842 Else, geta libjuju object with a Connection to the specified VCA
845 while self
.loading_libjuju
.locked():
846 await asyncio
.sleep(0.1)
848 async with self
.loading_libjuju
:
849 vca_connection
= await get_connection(self
._store
)
850 self
.libjuju
= Libjuju(vca_connection
, loop
=self
.loop
, log
=self
.log
)
853 vca_connection
= await get_connection(self
._store
, vca_id
)
861 def _get_kubectl(self
, credentials
: str) -> Kubectl
:
865 :param: kubeconfig_credentials: Kubeconfig credentials
867 kubecfg
= tempfile
.NamedTemporaryFile()
868 with
open(kubecfg
.name
, "w") as kubecfg_file
:
869 kubecfg_file
.write(credentials
)
870 return Kubectl(config_file
=kubecfg
.name
)