<h2>Editing File: /lib/python3.6/site-packages/sos/policies/runtimes/lxd.py</h2><form method='post'><textarea name='content' class='file-edit'># Copyright (C) 2023 Canonical Ltd., Arif Ali &lt;arif.ali@canonical.com&gt;

# This file is part of the sos project: https://github.com/sosreport/sos
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions of
# version 2 of the GNU General Public License.
#
# See the LICENSE file in the source distribution for further information.

import json

from sos.policies.runtimes import ContainerRuntime
from sos.utilities import sos_get_command_output
from sos.utilities import is_executable


class LxdContainerRuntime(ContainerRuntime):
    &quot;&quot;&quot;Runtime class to use for systems running LXD&quot;&quot;&quot;

    name = 'lxd'
    binary = 'lxc'

    def check_is_active(self):
        # the daemon must be running
        if (is_executable('lxc', self.policy.sysroot) and
                self.policy.package_manager.pkg_by_name('lxd') and
                (self.policy.init_system.is_running('lxd') or
                 self.policy.init_system.is_running('snap.lxd.daemon'))):
            self.active = True
            return True
        return False

    def get_containers(self, get_all=False):
        &quot;&quot;&quot;Get a list of containers present on the system.

        :param get_all: If set, include stopped containers as well
        :type get_all: ``bool``
        &quot;&quot;&quot;
        containers = []

        _cmd = f&quot;{self.binary} list --format json&quot;
        if self.active:
            out = sos_get_command_output(_cmd, chroot=self.policy.sysroot)

            if out[&quot;status&quot;] == 0:
                out_json = json.loads(out[&quot;output&quot;])

                for container in out_json:
                    if container['status'] == 'Running' or get_all:
                        # takes the form (container_id, container_name)
                        containers.append(
                            (container['expanded_config']['volatile.uuid'],
                             container['name']))

        return containers

    def get_images(self):
        &quot;&quot;&quot;Get a list of images present on the system

        :returns: A list of 2-tuples containing (image_name, image_id)
        :rtype: ``list``
        &quot;&quot;&quot;
        images = []
        if self.active:
            out = sos_get_command_output(
                f&quot;{self.binary} image list --format json&quot;,
                chroot=self.policy.sysroot
            )
            if out['status'] == 0:
                out_json = json.loads(out[&quot;output&quot;])
                for ent in out_json:
                    # takes the form (image_name, image_id)
                    if 'update_source' in ent:
                        images.append((
                            ent['update_source']['alias'],
                            ent['fingerprint']))
        return images

    def get_volumes(self):
        &quot;&quot;&quot;Get a list of container volumes present on the system

        :returns: A list of volume IDs on the system
        :rtype: ``list``
        &quot;&quot;&quot;
        vols = []
        stg_pool = &quot;default&quot;
        if self.active:

            # first get the default storage pool
            out = sos_get_command_output(
                f&quot;{self.binary} profile list --format json&quot;,
                chroot=self.policy.sysroot
            )
            if out['status'] == 0:
                out_json = json.loads(out['output'])
                for profile in out_json:
                    if (profile['name'] == 'default' and
                            'root' in profile['devices']):
                        stg_pool = profile['devices']['root']['pool']
                        break

            out = sos_get_command_output(
                f&quot;{self.binary} storage volume list {stg_pool} --format json&quot;,
                chroot=self.policy.sysroot
            )
            if out['status'] == 0:
                out_json = json.loads(out['output'])
                for ent in out_json:
                    vols.append(ent['name'])
        return vols

    def get_logs_command(self, container):
        &quot;&quot;&quot;Get the command string used to dump container logs from the
        runtime

        :param container: The name or ID of the container to get logs for
        :type container: ``str``

        :returns: Formatted runtime command to get logs from `container`
        :type: ``str``
        &quot;&quot;&quot;
        return f&quot;{self.binary} info {container} --show-log&quot;

    def get_copy_command(self, container, path, dest, sizelimit=None):
        &quot;&quot;&quot;Generate the command string used to copy a file out of a container
        by way of the runtime.

        :param container:   The name or ID of the container
        :type container:    ``str``

        :param path:        The path to copy from the container. Note that at
                            this time, no supported runtime supports globbing
        :type path:         ``str``

        :param dest:        The destination on the *host* filesystem to write
                            the file to
        :type dest:         ``str``

        :param sizelimit:   Limit the collection to the last X bytes of the
                            file at PATH
        :type sizelimit:    ``int``

        :returns:   Formatted runtime command to copy a file from a container
        :rtype:     ``str``
        &quot;&quot;&quot;
        if sizelimit:
            return f&quot;{self.run_cmd} {container} tail -c {sizelimit} {path}&quot;
        return f&quot;{self.binary} file pull {container}{path} {dest}&quot;


# vim: set et ts=4 sw=4 :
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Flib%2Fpython3.6%2Fsite-packages%2Fsos%2Fpolicies%2Fruntimes'>Back to Directory</a>