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
.config
import EnvironConfig
24 from n2vc
.exceptions
import K8sException
25 from n2vc
.k8s_conn
import K8sConnector
26 from n2vc
.kubectl
import Kubectl
, CORE_CLIENT
, RBAC_CLIENT
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
33 from kubernetes
.client
.models
import (
43 from typing
import Dict
45 SERVICE_ACCOUNT_TOKEN_KEY
= "token"
46 SERVICE_ACCOUNT_ROOT_CA_KEY
= "ca.crt"
47 RBAC_LABEL_KEY_NAME
= "rbac-id"
49 ADMIN_NAMESPACE
= "kube-system"
50 RBAC_STACK_PREFIX
= "juju-credential"
53 def generate_rbac_id():
54 return binascii
.hexlify(os
.urandom(4)).decode()
57 class K8sJujuConnector(K8sConnector
):
64 kubectl_command
: str = "/usr/bin/kubectl",
65 juju_command
: str = "/usr/bin/juju",
71 :param fs: file system for kubernetes and helm configuration
72 :param db: Database object
73 :param kubectl_command: path to kubectl executable
74 :param helm_command: path to helm executable
76 :param: loop: Asyncio loop
80 K8sConnector
.__init
__(
84 on_update_db
=on_update_db
,
88 self
.loop
= loop
or asyncio
.get_event_loop()
89 self
.log
.debug("Initializing K8S Juju connector")
91 db_uri
= EnvironConfig(prefixes
=["OSMLCM_", "OSMMON_"]).get("database_uri")
92 self
._store
= MotorStore(db_uri
)
93 self
.loading_libjuju
= asyncio
.Lock(loop
=self
.loop
)
95 self
.log
.debug("K8S Juju connector initialized")
96 # TODO: Remove these commented lines:
97 # self.authenticated = False
99 # self.juju_secret = ""
106 namespace
: str = "kube-system",
107 reuse_cluster_uuid
: str = None,
111 It prepares a given K8s cluster environment to run Juju bundles.
113 :param k8s_creds: credentials to access a given K8s cluster, i.e. a valid
115 :param namespace: optional namespace to be used for juju. By default,
116 'kube-system' will be used
117 :param reuse_cluster_uuid: existing cluster uuid for reuse
118 :param: kwargs: Additional parameters
121 :return: uuid of the K8s cluster and True if connector has installed some
122 software in the cluster
123 (on error, an exception will be raised)
125 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
127 cluster_uuid
= reuse_cluster_uuid
or str(uuid
.uuid4())
129 kubecfg
= tempfile
.NamedTemporaryFile()
130 with
open(kubecfg
.name
, "w") as kubecfg_file
:
131 kubecfg_file
.write(k8s_creds
)
132 kubectl
= Kubectl(config_file
=kubecfg
.name
)
134 # CREATING RESOURCES IN K8S
135 rbac_id
= generate_rbac_id()
136 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
137 labels
= {RBAC_STACK_PREFIX
: rbac_id
}
139 # Create cleanup dictionary to clean up created resources
140 # if it fails in the middle of the process
143 self
._create
_cluster
_role
(
150 "delete": self
._delete
_cluster
_role
,
151 "args": (kubectl
, metadata_name
),
155 self
._create
_service
_account
(
162 "delete": self
._delete
_service
_account
,
163 "args": (kubectl
, metadata_name
),
167 self
._create
_cluster
_role
_binding
(
174 "delete": self
._delete
_service
_account
,
175 "args": (kubectl
, metadata_name
),
178 token
, client_cert_data
= await self
._get
_secret
_data
(
183 default_storage_class
= kubectl
.get_default_storage_class()
184 await libjuju
.add_k8s(
188 client_cert_data
=client_cert_data
,
189 configuration
=kubectl
.configuration
,
190 storage_class
=default_storage_class
,
191 credential_name
=self
._get
_credential
_name
(cluster_uuid
),
193 return cluster_uuid
, True
194 except Exception as e
:
195 self
.log
.error("Error initializing k8scluster: {}".format(e
))
196 if len(cleanup_data
) > 0:
197 self
.log
.debug("Cleaning up created resources in k8s cluster...")
198 for item
in cleanup_data
:
199 delete_function
= item
["delete"]
200 delete_args
= item
["args"]
201 delete_function(*delete_args
)
202 self
.log
.debug("Cleanup finished")
205 """Repo Management"""
211 _type
: str = "charm",
213 raise MethodNotImplemented()
215 async def repo_list(self
):
216 raise MethodNotImplemented()
218 async def repo_remove(
222 raise MethodNotImplemented()
224 async def synchronize_repos(self
, cluster_uuid
: str, name
: str):
226 Returns None as currently add_repo is not implemented
236 uninstall_sw
: bool = False,
241 Resets the Kubernetes cluster by removing the model that represents it.
243 :param cluster_uuid str: The UUID of the cluster to reset
244 :param force: Force reset
245 :param uninstall_sw: Boolean to uninstall sw
246 :param: kwargs: Additional parameters
249 :return: Returns True if successful or raises an exception.
253 self
.log
.debug("[reset] Removing k8s cloud")
254 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
256 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
258 cloud_creds
= await libjuju
.get_cloud_credentials(cloud
)
260 await libjuju
.remove_cloud(cluster_uuid
)
262 kubecfg
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
264 kubecfg_file
= tempfile
.NamedTemporaryFile()
265 with
open(kubecfg_file
.name
, "w") as f
:
267 kubectl
= Kubectl(config_file
=kubecfg_file
.name
)
270 self
._delete
_cluster
_role
_binding
,
271 self
._delete
_service
_account
,
272 self
._delete
_cluster
_role
,
275 credential_attrs
= cloud_creds
[0].result
["attrs"]
276 if RBAC_LABEL_KEY_NAME
in credential_attrs
:
277 rbac_id
= credential_attrs
[RBAC_LABEL_KEY_NAME
]
278 metadata_name
= "{}-{}".format(RBAC_STACK_PREFIX
, rbac_id
)
279 delete_args
= (kubectl
, metadata_name
)
280 for delete_func
in delete_functions
:
282 delete_func(*delete_args
)
283 except Exception as e
:
284 self
.log
.warning("Cannot remove resource in K8s {}".format(e
))
286 except Exception as e
:
287 self
.log
.debug("Caught exception during reset: {}".format(e
))
299 timeout
: float = 1800,
301 db_dict
: dict = None,
302 kdu_name
: str = None,
303 namespace
: str = None,
308 :param cluster_uuid str: The UUID of the cluster to install to
309 :param kdu_model str: The name or path of a bundle to install
310 :param kdu_instance: Kdu instance name
311 :param atomic bool: If set, waits until the model is active and resets
312 the cluster on failure.
313 :param timeout int: The time, in seconds, to wait for the install
315 :param params dict: Key-value pairs of instantiation parameters
316 :param kdu_name: Name of the KDU instance to be installed
317 :param namespace: K8s namespace to use for the KDU instance
318 :param kwargs: Additional parameters
321 :return: If successful, returns ?
323 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
327 raise K8sException("db_dict must be set")
329 raise K8sException("bundle must be set")
331 if bundle
.startswith("cs:"):
333 elif bundle
.startswith("http"):
337 new_workdir
= kdu_model
.strip(kdu_model
.split("/")[-1])
338 os
.chdir(new_workdir
)
339 bundle
= "local:{}".format(kdu_model
)
341 self
.log
.debug("Checking for model named {}".format(kdu_instance
))
343 # Create the new model
344 self
.log
.debug("Adding model: {}".format(kdu_instance
))
345 cloud
= Cloud(cluster_uuid
, self
._get
_credential
_name
(cluster_uuid
))
346 await libjuju
.add_model(kdu_instance
, cloud
)
349 # TODO: Instantiation parameters
352 "Juju bundle that models the KDU, in any of the following ways:
353 - <juju-repo>/<juju-bundle>
354 - <juju-bundle folder under k8s_models folder in the package>
355 - <juju-bundle tgz file (w/ or w/o extension) under k8s_models folder
357 - <URL_where_to_fetch_juju_bundle>
360 previous_workdir
= os
.getcwd()
361 except FileNotFoundError
:
362 previous_workdir
= "/app/storage"
364 self
.log
.debug("[install] deploying {}".format(bundle
))
365 await libjuju
.deploy(
366 bundle
, model_name
=kdu_instance
, wait
=atomic
, timeout
=timeout
368 os
.chdir(previous_workdir
)
369 if self
.on_update_db
:
370 await self
.on_update_db(
373 filter=db_dict
["filter"],
374 vca_id
=kwargs
.get("vca_id")
383 total_timeout
: float = 1800,
386 """Scale an application in a model
388 :param: kdu_instance str: KDU instance name
389 :param: scale int: Scale to which to set this application
390 :param: resource_name str: Resource name (Application name)
391 :param: timeout float: The time, in seconds, to wait for the install
393 :param kwargs: Additional parameters
396 :return: If successful, returns True
400 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
401 await libjuju
.scale_application(
402 model_name
=kdu_instance
,
403 application_name
=resource_name
,
405 total_timeout
=total_timeout
407 except Exception as e
:
408 error_msg
= "Error scaling application {} in kdu instance {}: {}".format(
409 resource_name
, kdu_instance
, e
)
410 self
.log
.error(error_msg
)
411 raise K8sException(message
=error_msg
)
414 async def get_scale_count(
420 """Get an application scale count
422 :param: resource_name str: Resource name (Application name)
423 :param: kdu_instance str: KDU instance name
424 :param kwargs: Additional parameters
426 :return: Return application instance count
429 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
430 status
= await libjuju
.get_model_status(kdu_instance
)
431 return len(status
.applications
[resource_name
].units
)
432 except Exception as e
:
433 error_msg
= "Error getting scale count from application {} in kdu instance {}: {}".format(
434 resource_name
, kdu_instance
, e
)
435 self
.log
.error(error_msg
)
436 raise K8sException(message
=error_msg
)
438 async def instances_list(self
, cluster_uuid
: str) -> list:
440 returns a list of deployed releases in a cluster
442 :param cluster_uuid: the cluster
451 kdu_model
: str = None,
456 :param cluster_uuid str: The UUID of the cluster to upgrade
457 :param kdu_instance str: The unique name of the KDU instance
458 :param kdu_model str: The name or path of the bundle to upgrade to
459 :param params dict: Key-value pairs of instantiation parameters
461 :return: If successful, reference to the new revision number of the
465 # TODO: Loop through the bundle and upgrade each charm individually
468 The API doesn't have a concept of bundle upgrades, because there are
469 many possible changes: charm revision, disk, number of units, etc.
471 As such, we are only supporting a limited subset of upgrades. We'll
472 upgrade the charm revision but leave storage and scale untouched.
474 Scale changes should happen through OSM constructs, and changes to
475 storage would require a redeployment of the service, at least in this
478 raise MethodNotImplemented()
490 :param cluster_uuid str: The UUID of the cluster to rollback
491 :param kdu_instance str: The unique name of the KDU instance
492 :param revision int: The revision to revert to. If omitted, rolls back
493 the previous upgrade.
495 :return: If successful, returns the revision of active KDU instance,
496 or raises an exception
498 raise MethodNotImplemented()
508 """Uninstall a KDU instance
510 :param cluster_uuid str: The UUID of the cluster
511 :param kdu_instance str: The unique name of the KDU instance
512 :param kwargs: Additional parameters
515 :return: Returns True if successful, or raises an exception
518 self
.log
.debug("[uninstall] Destroying model")
519 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
521 await libjuju
.destroy_model(kdu_instance
, total_timeout
=3600)
523 # self.log.debug("[uninstall] Model destroyed and disconnecting")
524 # await controller.disconnect()
527 # TODO: Remove these commented lines
528 # if not self.authenticated:
529 # self.log.debug("[uninstall] Connecting to controller")
530 # await self.login(cluster_uuid)
532 async def exec_primitive(
534 cluster_uuid
: str = None,
535 kdu_instance
: str = None,
536 primitive_name
: str = None,
537 timeout
: float = 300,
539 db_dict
: dict = None,
542 """Exec primitive (Juju action)
544 :param cluster_uuid str: The UUID of the cluster
545 :param kdu_instance str: The unique name of the KDU instance
546 :param primitive_name: Name of action that will be executed
547 :param timeout: Timeout for action execution
548 :param params: Dictionary of all the parameters needed for the action
549 :param db_dict: Dictionary for any additional data
550 :param kwargs: Additional parameters
553 :return: Returns the output of the action
555 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
557 if not params
or "application-name" not in params
:
559 "Missing application-name argument, \
560 argument needed for K8s actions"
564 "[exec_primitive] Getting model "
565 "kdu_instance: {}".format(kdu_instance
)
567 application_name
= params
["application-name"]
568 actions
= await libjuju
.get_actions(application_name
, kdu_instance
)
569 if primitive_name
not in actions
:
570 raise K8sException("Primitive {} not found".format(primitive_name
))
571 output
, status
= await libjuju
.execute_action(
572 application_name
, kdu_instance
, primitive_name
, **params
575 if status
!= "completed":
577 "status is not completed: {} output: {}".format(status
, output
)
579 if self
.on_update_db
:
580 await self
.on_update_db(cluster_uuid
, kdu_instance
, filter=db_dict
["filter"])
584 except Exception as e
:
585 error_msg
= "Error executing primitive {}: {}".format(primitive_name
, e
)
586 self
.log
.error(error_msg
)
587 raise K8sException(message
=error_msg
)
591 async def inspect_kdu(
597 Inspects a bundle and returns a dictionary of config parameters and
598 their default values.
600 :param kdu_model str: The name or path of the bundle to inspect.
602 :return: If successful, returns a dictionary of available parameters
603 and their default values.
607 if not os
.path
.exists(kdu_model
):
608 raise K8sException("file {} not found".format(kdu_model
))
610 with
open(kdu_model
, "r") as f
:
611 bundle
= yaml
.safe_load(f
.read())
615 'description': 'Test bundle',
616 'bundle': 'kubernetes',
619 'charm': 'cs:~charmed-osm/mariadb-k8s-20',
622 'password': 'manopw',
623 'root_password': 'osm4u',
626 'series': 'kubernetes'
631 # TODO: This should be returned in an agreed-upon format
632 kdu
= bundle
["applications"]
642 If available, returns the README of the bundle.
644 :param kdu_model str: The name or path of a bundle
646 :return: If found, returns the contents of the README.
650 files
= ["README", "README.txt", "README.md"]
651 path
= os
.path
.dirname(kdu_model
)
652 for file in os
.listdir(path
):
654 with
open(file, "r") as f
:
660 async def status_kdu(
664 complete_status
: bool = False,
665 yaml_format
: bool = False,
668 """Get the status of the KDU
670 Get the current status of the KDU instance.
672 :param cluster_uuid str: The UUID of the cluster
673 :param kdu_instance str: The unique id of the KDU instance
674 :param complete_status: To get the complete_status of the KDU
675 :param yaml_format: To get the status in proper format for NSR record
676 :param: kwargs: Additional parameters
679 :return: Returns a dictionary containing namespace, state, resources,
680 and deployment_time and returns complete_status if complete_status is True
682 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
685 model_status
= await libjuju
.get_model_status(kdu_instance
)
687 if not complete_status
:
688 for name
in model_status
.applications
:
689 application
= model_status
.applications
[name
]
690 status
[name
] = {"status": application
["status"]["status"]}
693 return obj_to_yaml(model_status
)
695 return obj_to_dict(model_status
)
699 async def update_vca_status(self
, vcastatus
: dict, kdu_instance
: str, **kwargs
):
701 Add all configs, actions, executed actions of all applications in a model to vcastatus dict
703 :param vcastatus dict: dict containing vcastatus
704 :param kdu_instance str: The unique id of the KDU instance
705 :param: kwargs: Additional parameters
710 libjuju
= await self
._get
_libjuju
(kwargs
.get("vca_id"))
712 for model_name
in vcastatus
:
713 # Adding executed actions
714 vcastatus
[model_name
]["executedActions"] = \
715 await libjuju
.get_executed_actions(kdu_instance
)
717 for application
in vcastatus
[model_name
]["applications"]:
718 # Adding application actions
719 vcastatus
[model_name
]["applications"][application
]["actions"] = \
720 await libjuju
.get_actions(application
, kdu_instance
)
721 # Adding application configs
722 vcastatus
[model_name
]["applications"][application
]["configs"] = \
723 await libjuju
.get_application_configs(kdu_instance
, application
)
725 except Exception as e
:
726 self
.log
.debug("Error in updating vca status: {}".format(str(e
)))
728 async def get_services(
729 self
, cluster_uuid
: str, kdu_instance
: str, namespace
: str
731 """Return a list of services of a kdu_instance"""
733 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
735 kubecfg
= tempfile
.NamedTemporaryFile()
736 with
open(kubecfg
.name
, "w") as kubecfg_file
:
737 kubecfg_file
.write(credentials
)
738 kubectl
= Kubectl(config_file
=kubecfg
.name
)
740 return kubectl
.get_services(
741 field_selector
="metadata.namespace={}".format(kdu_instance
)
744 async def get_service(
745 self
, cluster_uuid
: str, service_name
: str, namespace
: str
747 """Return data for a specific service inside a namespace"""
749 credentials
= self
.get_credentials(cluster_uuid
=cluster_uuid
)
751 kubecfg
= tempfile
.NamedTemporaryFile()
752 with
open(kubecfg
.name
, "w") as kubecfg_file
:
753 kubecfg_file
.write(credentials
)
754 kubectl
= Kubectl(config_file
=kubecfg
.name
)
756 return kubectl
.get_services(
757 field_selector
="metadata.name={},metadata.namespace={}".format(
758 service_name
, namespace
762 def get_credentials(self
, cluster_uuid
: str) -> str:
764 Get Cluster Kubeconfig
766 k8scluster
= self
.db
.get_one(
767 "k8sclusters", q_filter
={"_id": cluster_uuid
}, fail_on_empty
=False
770 self
.db
.encrypt_decrypt_fields(
771 k8scluster
.get("credentials"),
773 ["password", "secret"],
774 schema_version
=k8scluster
["schema_version"],
775 salt
=k8scluster
["_id"],
778 return yaml
.safe_dump(k8scluster
.get("credentials"))
780 def _get_credential_name(self
, cluster_uuid
: str) -> str:
782 Get credential name for a k8s cloud
784 We cannot use the cluster_uuid for the credential name directly,
785 because it cannot start with a number, it must start with a letter.
786 Therefore, the k8s cloud credential name will be "cred-" followed
789 :param: cluster_uuid: Cluster UUID of the kubernetes cloud (=cloud_name)
791 :return: Name to use for the credential name.
793 return "cred-{}".format(cluster_uuid
)
799 """Get the namespace UUID
800 Gets the namespace's unique name
802 :param cluster_uuid str: The UUID of the cluster
803 :returns: The namespace UUID, or raises an exception
807 def _create_cluster_role(
811 labels
: Dict
[str, str],
813 cluster_roles
= kubectl
.clients
[RBAC_CLIENT
].list_cluster_role(
814 field_selector
="metadata.name={}".format(name
)
817 if len(cluster_roles
.items
) > 0:
819 "Cluster role with metadata.name={} already exists".format(name
)
822 metadata
= V1ObjectMeta(name
=name
, labels
=labels
, namespace
=ADMIN_NAMESPACE
)
824 cluster_role
= V1ClusterRole(
827 V1PolicyRule(api_groups
=["*"], resources
=["*"], verbs
=["*"]),
828 V1PolicyRule(non_resource_ur_ls
=["*"], verbs
=["*"]),
832 kubectl
.clients
[RBAC_CLIENT
].create_cluster_role(cluster_role
)
834 def _delete_cluster_role(self
, kubectl
: Kubectl
, name
: str):
835 kubectl
.clients
[RBAC_CLIENT
].delete_cluster_role(name
)
837 def _create_service_account(
841 labels
: Dict
[str, str],
843 service_accounts
= kubectl
.clients
[CORE_CLIENT
].list_namespaced_service_account(
844 ADMIN_NAMESPACE
, field_selector
="metadata.name={}".format(name
)
846 if len(service_accounts
.items
) > 0:
848 "Service account with metadata.name={} already exists".format(name
)
851 metadata
= V1ObjectMeta(name
=name
, labels
=labels
, namespace
=ADMIN_NAMESPACE
)
852 service_account
= V1ServiceAccount(metadata
=metadata
)
854 kubectl
.clients
[CORE_CLIENT
].create_namespaced_service_account(
855 ADMIN_NAMESPACE
, service_account
858 def _delete_service_account(self
, kubectl
: Kubectl
, name
: str):
859 kubectl
.clients
[CORE_CLIENT
].delete_namespaced_service_account(
860 name
, ADMIN_NAMESPACE
863 def _create_cluster_role_binding(
867 labels
: Dict
[str, str],
869 role_bindings
= kubectl
.clients
[RBAC_CLIENT
].list_cluster_role_binding(
870 field_selector
="metadata.name={}".format(name
)
872 if len(role_bindings
.items
) > 0:
873 raise Exception("Generated rbac id already exists")
875 role_binding
= V1ClusterRoleBinding(
876 metadata
=V1ObjectMeta(name
=name
, labels
=labels
),
877 role_ref
=V1RoleRef(kind
="ClusterRole", name
=name
, api_group
=""),
879 V1Subject(kind
="ServiceAccount", name
=name
, namespace
=ADMIN_NAMESPACE
)
882 kubectl
.clients
[RBAC_CLIENT
].create_cluster_role_binding(role_binding
)
884 def _delete_cluster_role_binding(self
, kubectl
: Kubectl
, name
: str):
885 kubectl
.clients
[RBAC_CLIENT
].delete_cluster_role_binding(name
)
887 async def _get_secret_data(self
, kubectl
: Kubectl
, name
: str) -> (str, str):
888 v1_core
= kubectl
.clients
[CORE_CLIENT
]
894 service_accounts
= v1_core
.list_namespaced_service_account(
895 ADMIN_NAMESPACE
, field_selector
="metadata.name={}".format(name
)
897 if len(service_accounts
.items
) == 0:
899 "Service account not found with metadata.name={}".format(name
)
901 service_account
= service_accounts
.items
[0]
902 if service_account
.secrets
and len(service_account
.secrets
) > 0:
903 secret_name
= service_account
.secrets
[0].name
904 if secret_name
is not None or not retries_limit
:
908 "Failed getting the secret from service account {}".format(name
)
910 secret
= v1_core
.list_namespaced_secret(
912 field_selector
="metadata.name={}".format(secret_name
),
915 token
= secret
.data
[SERVICE_ACCOUNT_TOKEN_KEY
]
916 client_certificate_data
= secret
.data
[SERVICE_ACCOUNT_ROOT_CA_KEY
]
919 base64
.b64decode(token
).decode("utf-8"),
920 base64
.b64decode(client_certificate_data
).decode("utf-8"),
924 def generate_kdu_instance_name(**kwargs
):
925 db_dict
= kwargs
.get("db_dict")
926 kdu_name
= kwargs
.get("kdu_name", None)
928 kdu_instance
= "{}-{}".format(kdu_name
, db_dict
["filter"]["_id"])
930 kdu_instance
= db_dict
["filter"]["_id"]
933 async def _get_libjuju(self
, vca_id
: str = None) -> Libjuju
:
937 :param: vca_id: VCA ID
938 If None, get a libjuju object with a Connection to the default VCA
939 Else, geta libjuju object with a Connection to the specified VCA
942 while self
.loading_libjuju
.locked():
943 await asyncio
.sleep(0.1)
945 async with self
.loading_libjuju
:
946 vca_connection
= await get_connection(self
._store
)
947 self
.libjuju
= Libjuju(vca_connection
, loop
=self
.loop
, log
=self
.log
)
950 vca_connection
= await get_connection(self
._store
, vca_id
)