[JUJU-3894] Forward port some fixes from 2.9 to master (#870)

* Add kubernetes as supported series as per juju/core/series/supported.go

Fixes #865

* Add example local charm to test and reproduce #865

Unfortunately we can't write an integration test that uses this
because we run our tests on lxd so the charm will be deployed but will
never actually get 'active'. We could test the deployment itself (that
it doesn't error), but if we did that, the 3.0 will actually fail to
deploy on lxd because the series is kubernetes so the test will be
invalid anyways.

* Fix bug in Type.from_json() parsing simple entries

Should fix #850 & #851

* Add integration test for simple assumes expression

This is not the ideal test because it depends on that the upstream
charm having the simple

assumes:
  -juju

expression. However, simulating the bug in the facade.py for parsing
such expressions is non-trivial as we need a test to call something
like the CharmsFacade.CharmInfo() to trigger parsing the metadata
(which is actually where it fails in the reported bug). Creating a
local charm wouldn't work because we locally handle the metadata
(wihtout going through anything in the facade.py where the bug is
located). Maybe we can manually call AddCharm in the test for a local
charm and then manually call the CharmInfo with the returned url.

* Fix wait_for_units flag to not block when enough units ready

wait_for_idle will keep waiting if there are less number of units
available than requested (via the wait_for_units flag). However, if
there are already a number of units in a desired status ready to go,
more than (or equal to) wait_for_units, then it shouldn't block until
other not-yet-available units to get into that desired state as well.

fixes #837

* Add integration test for wait_for_units in wait_for_idle

* Fix failing wait_for_idle test

As per discussion in
#841 (comment)

Should fix #837

* Remove accidental printf for debugging

* Small patch for wait_for_idle

* Fix wait_for_exat_units=0 case

* Fix logical typo

* Fix merge resolve error for parsing assumes

* Fix base channel discovery for local charms

`utils.get_local_charm_base()` was incorrectly using the `--channel`
argument (the charm's channel) for discovering the channel part of the
base. (we should stop using the word 'channel' for two different
things).

This fixes that by taking out the incorrect part of the code.

Should fix #839

* Fixes to pass the CI problems regarding missing postgresql charm. (#847)

* Add test for deploying local charm with channel argument

* Add application.get_status to get the most up to date status from API

Introduces an internal self._status which is initially set to the
'unknown' which has the lowest severity. The regular property
self.status uses both self._status and the unit statuses to derive the
most severe status as the application status

* Use application.get_status in wait_for_idle to use most up to date
application status

* Fix unit test TestModelWaitForIdle::test_wait_for_active_status

* Fix linter

* Expect and handle exceptions from the AllWatcher task

fixes #829

The `_all_watcher` task is a coroutine for the AllWatcher to run in
the background all the time forever, and it involves a while loop
that's being controlled manually through some flags (asyncio events),
e.g. things like `_watch_stopping`, `watch_stopped`.

The problem is that when the `_all_watcher` raises an exception (or
receives one from things like `get_config()` like in the case of
ether in the event loop, not handled/or re-raised. This is because
this coroutine is not `await`ed (for good reason), it can't be
`await`ed because there won't ever be any results, this method is
supposed to be working in the background forever getting the deltas
for us. As a result of this, if `_all_watcher` fails, then external
flags like `_watch_received` is never set, and whoever's calling
`await self._watch_received.wait()` will block forever (in this case
the `_after_connect()`. Similarly the `disconnect()` waits for the
`_watch_stopped` flag, which won't be set either, so if we call
disconnect when all_watcher failed then it'll hang forever.

This change fixes this problem by allowing (at the wait-for-flag
spots) to wait for two things, 1) whichever flag we're waiting for, 2)
`_all_watcher` task to be `"done()"`. In the latter case, we should
expect to see an exception because that task is not supposed to be
finished. More importantly, if we do see that the
`_watcher_task.done()`, then we don't sit and wait forever for the
_all_watcher event flags to be set, so we won't hang.

Also a nice side effect of this should be that we should be getting
less number of extra exception outputs saying that the "Task exception
is never handled", since we do call the `.exception()` on the
`_all_watcher` task. Though we'll probably continue to get those from
the tasks like `_pinger` and `_debug_log` etc. However, this is a good
first example solution to handle them as well.

* Assume ubuntu focal base for legacy k8s charms

Updates the get_local_charm_base with Base(20.04/stable, ubuntu) for
legacy k8s charms as per
juju/cmd/juju/application/utils.DeduceOrigin()

* Fix get_local_charm_base call.

---------

Co-authored-by: Juan M. Tirado <juanmanuel-tirado@users.noreply.github.com>
Co-authored-by: Juan Tirado <juan.tirado@canonical.com>

Author: 3 people

juju/application.py

@@ -91,10 +91,15 @@ def status(self):
         """
         status = self.safe_data['status']['current']
         if status == "unset":
-            unit_status = []
+            known_statuses = []
             for unit in self.units:
-                unit_status.append(unit.workload_status)
-            return derive_status(unit_status)
+                known_statuses.append(unit.workload_status)
+            # If the self.get_status() is called (i.e. the status
+            # is received by FullStatus from the API) then add
+            # that into this computation as it might be more up
+            # to date (and more severe).
+            known_statuses.append(self._status)
+            return derive_status(known_statuses)
         return status
 
     @property
@@ -465,6 +470,23 @@ async def get_actions(self, schema=False):
             actions = {k: v.description for k, v in actions.items()}
         return actions
 
+    async def get_status(self):
+        """Get the application status using info from the FullStatus
+        as well, because it might be more up to date than our model
+
+        :return: str status
+        """
+
+        client_facade = client.ClientFacade.from_connection(self.connection)
+
+        full_status = await client_facade.FullStatus(patterns=None)
+        _app = full_status.applications.get(self.name, None)
+        if not _app:
+            raise JujuError(f"application is not in FullStatus : {self.name}")
+
+        self._status = derive_status([self.status, _app.status.status])
+        return self._status
+
     def attach_resource(self, resource_name, file_name, file_obj):
         """Updates the resource for an application by uploading file from
         local disk to the Juju controller.

juju/bundle.py

@@ -27,6 +27,7 @@ class BundleHandler:
     Handle bundles by using the API to translate bundle YAML into a plan of
     steps and then dispatching each of those using the API.
     """
+
     def __init__(self, model, trusted=False, forced=False):
         self.model = model
         self.trusted = trusted
@@ -159,8 +160,7 @@ async def _handle_local_charms(self, bundle, bundle_dir):
                 apps_dict[app_name]["resources"] = resources
                 origin = client.CharmOrigin(source="local", risk="stable")
                 if not self.model.connection().is_using_old_client:
-                    origin.base = utils.get_local_charm_base(series, '',
-                                                             metadata,
+                    origin.base = utils.get_local_charm_base(series,
                                                              charm_dir,
                                                              client.Base)
                 self.origins[charm_url] = {str(None): origin}
@@ -523,6 +523,7 @@ class AddApplicationChange(ChangeInfo):
     :local_resources: identifies the path to the local resource of the
         application's charm.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(AddApplicationChange, self).__init__(change_id, requires)
 
@@ -669,6 +670,7 @@ class AddCharmChange(ChangeInfo):
            not sufficient.
         :channel: preferred channel for obtaining the charm.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(AddCharmChange, self).__init__(change_id, requires)
 
@@ -770,6 +772,7 @@ class AddMachineChange(ChangeInfo):
             "lxc" or kvm"). It is not specified for top level machines.
         :parent_id: id of the parent machine.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(AddMachineChange, self).__init__(change_id, requires)
         # this one is weird, as it returns a set of parameters inside a list.
@@ -861,6 +864,7 @@ class AddRelationChange(ChangeInfo):
         application, and the interface is optional. Examples are
         "$deploy-42:web", "$deploy-42", "mysql:db".
     """
+
     def __init__(self, change_id, requires, params=None):
         super(AddRelationChange, self).__init__(change_id, requires)
 
@@ -922,6 +926,7 @@ class AddUnitChange(ChangeInfo):
         :to: optional location where to add the unit, as a placeholder
             pointing to another unit change or to a machine change.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(AddUnitChange, self).__init__(change_id, requires)
 
@@ -990,6 +995,7 @@ class CreateOfferChange(ChangeInfo):
             offer.
         :offer_name: describes the offer name.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(CreateOfferChange, self).__init__(change_id, requires)
 
@@ -1044,6 +1050,7 @@ class ConsumeOfferChange(ChangeInfo):
         :url: contains the location of the offer
         :application_name: describes the application name on offer.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(ConsumeOfferChange, self).__init__(change_id, requires)
 
@@ -1099,6 +1106,7 @@ class ExposeChange(ChangeInfo):
             that should be able to access the port ranges that the application
             has opened for each endpoint.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(ExposeChange, self).__init__(change_id, requires)
 
@@ -1148,6 +1156,7 @@ class ScaleChange(ChangeInfo):
         :application: placeholder name of the application to be scaled.
         :scale: is the new scale value to use.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(ScaleChange, self).__init__(change_id, requires)
 
@@ -1200,6 +1209,7 @@ class SetAnnotationsChange(ChangeInfo):
         :entity_type: type of the entity, "application" or "machine".
         :ennotations: annotations as key/value pairs.
     """
+
     def __init__(self, change_id, requires, params=None):
         super(SetAnnotationsChange, self).__init__(change_id, requires)
 

juju/client/facade.py

@@ -685,7 +685,7 @@ def from_json(cls, data):
                     value = entry[i:].strip()
                     d[key] = value
                 else:
-                    # something like k8s-api
+                    # this is a simple entry
                     d[entry] = ''
             return cls(**d)
         return None

juju/model.py

@@ -262,6 +262,7 @@ def __init__(self, entity_id, model, history_index=-1, connected=True):
         self._history_index = history_index
         self.connected = connected
         self.connection = model.connection()
+        self._status = 'unknown'
 
     def __repr__(self):
         return '<{} entity_id="{}">'.format(type(self).__name__,
@@ -740,7 +741,26 @@ async def _after_connect(self, model_name=None, model_uuid=None):
         # we've received all the model data, which might be
         # a whole load of unneeded data if all the client wants
         # to do is make one RPC call.
-        await self._watch_received.wait()
+        async def watch_received_waiter():
+            await self._watch_received.wait()
+        waiter = jasyncio.create_task(watch_received_waiter())
+
+        # If we just wait for the _watch_received event and the _all_watcher task
+        # fails (e.g. because API fails like migration is in progress), then
+        # we'll hang because the _watch_received will never be set
+        # Instead, we watch for two things, 1) _watch_received, 2) _all_watcher done
+        # If _all_watcher is done before the _watch_received, then we should see
+        # (and raise) an exception coming from the _all_watcher
+        # Otherwise (i.e. _watch_received is set), then we're good to go
+        done, pending = await jasyncio.wait([waiter, self._watcher_task],
+                                            return_when=jasyncio.FIRST_COMPLETED)
+        if self._watcher_task in done:
+            # Cancel the _watch_received.wait
+            waiter.cancel()
+            # If there's no exception, then why did the _all_watcher broke its loop?
+            if not self._watcher_task.exception():
+                raise JujuError("AllWatcher task is finished abruptly without an exception.")
+            raise self._watcher_task.exception()
 
         if self.info is None:
             contr = await self.get_controller()
@@ -756,6 +776,12 @@ async def disconnect(self):
         if not self._watch_stopped.is_set():
             log.debug('Stopping watcher task')
             self._watch_stopping.set()
+            # If the _all_watcher task is finished,
+            # check to see an exception, if yes, raise,
+            # otherwise we should see the _watch_stopped
+            # flag is set
+            if self._watcher_task.done() and self._watcher_task.exception():
+                raise self._watcher_task.exception()
             await self._watch_stopped.wait()
             self._watch_stopping.clear()
 
@@ -1133,6 +1159,7 @@ def _watch(self):
         See :meth:`add_observer` to register an onchange callback.
 
         """
+
         def _post_step(obj):
             # Once we get the model, ensure we're running in the correct state
             # as a post step.
@@ -1222,7 +1249,7 @@ async def _all_watcher():
         self._watch_received.clear()
         self._watch_stopping.clear()
         self._watch_stopped.clear()
-        jasyncio.ensure_future(_all_watcher())
+        self._watcher_task = jasyncio.create_task(_all_watcher())
 
     async def _notify_observers(self, delta, old_obj, new_obj):
         """Call observing callbacks, notifying them of a change in model state
@@ -1744,11 +1771,10 @@ async def deploy(
                 # We have a local charm dir that needs to be uploaded
                 charm_dir = os.path.abspath(os.path.expanduser(identifier))
                 metadata = utils.get_local_charm_metadata(charm_dir)
-                charm_series = charm_series or await get_charm_series(metadata,
-                                                                      self)
-                if not base:
-                    charm_origin.base = utils.get_local_charm_base(
-                        charm_series, channel, metadata, charm_dir, client.Base)
+                charm_series = charm_series or await get_charm_series(metadata, self)
+
+                base = utils.get_local_charm_base(charm_series, charm_dir, client.Base)
+                charm_origin.base = base
 
                 if not application_name:
                     application_name = metadata['name']
@@ -2497,7 +2523,7 @@ async def _get_source_api(self, url, controller_name=None):
 
     async def wait_for_idle(self, apps=None, raise_on_error=True, raise_on_blocked=False,
                             wait_for_active=False, timeout=10 * 60, idle_period=15, check_freq=0.5,
-                            status=None, wait_for_units=1, wait_for_exact_units=None):
+                            status=None, wait_for_units=None, wait_for_exact_units=None):
         """Wait for applications in the model to settle into an idle state.
 
         :param apps (list[str]): Optional list of specific app names to wait on.
@@ -2545,6 +2571,8 @@ async def wait_for_idle(self, apps=None, raise_on_error=True, raise_on_blocked=F
             warnings.warn("wait_for_active is deprecated; use status", DeprecationWarning)
             status = "active"
 
+        _wait_for_units = wait_for_units if wait_for_units is not None else 1
+
         timeout = timedelta(seconds=timeout) if timeout is not None else None
         idle_period = timedelta(seconds=idle_period)
         start_time = datetime.now()
@@ -2556,6 +2584,7 @@ async def wait_for_idle(self, apps=None, raise_on_error=True, raise_on_blocked=F
 
         apps = apps or self.applications
         idle_times = {}
+        units_ready = set()  # The units that are in the desired state
         last_log_time = None
         log_interval = timedelta(seconds=30)
 
@@ -2581,6 +2610,9 @@ def _raise_for_status(entities, status):
                 'Invalid value for wait_for_exact_units : %s' % wait_for_exact_units
 
         while True:
+            # The list 'busy' is what keeps this loop going,
+            # i.e. it'll stop when busy is empty after all the
+            # units are scanned
             busy = []
             errors = {}
             blocks = {}
@@ -2589,19 +2621,28 @@ def _raise_for_status(entities, status):
                     busy.append(app_name + " (missing)")
                     continue
                 app = self.applications[app_name]
-                if raise_on_error and app.status == "error":
+                app_status = await app.get_status()
+                if raise_on_error and app_status == "error":
                     errors.setdefault("App", []).append(app.name)
-                if raise_on_blocked and app.status == "blocked":
+                if raise_on_blocked and app_status == "blocked":
                     blocks.setdefault("App", []).append(app.name)
+
+                # Check if wait_for_exact_units flag is used
                 if wait_for_exact_units is not None:
                     if len(app.units) != wait_for_exact_units:
                         busy.append(app.name + " (waiting for exactly %s units, current : %s)" %
                                     (wait_for_exact_units, len(app.units)))
                         continue
-                elif len(app.units) < wait_for_units:
+                # If we have less # of units then required, then wait a bit more
+                elif len(app.units) < _wait_for_units:
                     busy.append(app.name + " (not enough units yet - %s/%s)" %
-                                (len(app.units), wait_for_units))
+                                (len(app.units), _wait_for_units))
                     continue
+                # User wants to see a certain # of units, and we have enough
+                elif wait_for_units and len(units_ready) >= _wait_for_units:
+                    # So no need to keep looking, we have the desired number of units ready to go,
+                    # exit the loop. Don't return, though, we might still have some errors to raise
+                    break
                 for unit in app.units:
                     if unit.machine is not None and unit.machine.status == "error":
                         errors.setdefault("Machine", []).append(unit.machine.id)
@@ -2615,10 +2656,17 @@ def _raise_for_status(entities, status):
                     if raise_on_blocked and unit.workload_status == "blocked":
                         blocks.setdefault("Unit", []).append(unit.name)
                         continue
-                    waiting_for_status = status and unit.workload_status != status
-                    if not waiting_for_status and unit.agent_status == "idle":
+                    waiting_for_a_particular_status = status and unit.workload_status != status
+                    if not waiting_for_a_particular_status and unit.agent_status == "idle":
+                        # We'll be here in two cases:
+                        # 1) We're not waiting for a particular status and the agent is "idle"
+                        # 2) We're waiting for a particular status and the workload is in that status
+                        # Either way, the unit is ready, start measuring the time period that
+                        # it needs to stay in that state (i.e. idle_period)
+                        units_ready.add(unit.name)
                         now = datetime.now()
                         idle_start = idle_times.setdefault(unit.name, now)
+
                         if now - idle_start < idle_period:
                             busy.append("{} [{}] {}: {}".format(unit.name,
                                                                 unit.agent_status,