How to use the eliot.MessageType function in eliot

``IDeployer`` being tested.
    :param Leases leases: Currently configured leases. By default none exist.
    """
    cluster_state = compute_cluster_state(node_state, additional_node_states,
                                          nonmanifest_datasets)
    cluster_configuration = Deployment(
        nodes={node_config} | additional_node_config,
        leases=leases,
    )
    changes = deployer.calculate_changes(
        cluster_configuration, cluster_state, local_state
    )
    case.assertEqual(expected_changes, changes)


ADDRESS_IN_USE = MessageType(
    u"flocker:test:address_in_use",
    fields(ip=unicode, port=int, name=bytes),
)


def _find_process_name(port_number):
    """
    Get the name of the process using the given port number.
    """
    for connection in psutil.net_connections():
        if connection.laddr[1] == port_number:
            return psutil.Process(connection.pid).name()
    return None


def _retry_on_port_collision(reason, add, cleanup):

return self

    def commit(self):
        return self._root.commit()


TARGET_OBJECT = Field(
    u"target_object", repr,
    u"The object to which the diff was applied."
)
CHANGES = Field(
    u"changes", repr,
    u"The changes being applied."
)

DIFF_COMMIT_ERROR = MessageType(
    u"flocker:control:Diff:commit_error",
    [TARGET_OBJECT, CHANGES],
    u"The target and changes that failed to apply."
)


@implementer(_IDiffChange)
class Diff(PClass):
    """
    A ``_IDiffChange`` that is simply the serial application of other diff
    changes.

    This is the object that external modules get and use to apply diffs to
    objects.

    :ivar changes: A vector of ``_IDiffChange`` s that represent a diff between

from twisted.internet.error import ProcessTerminated, ProcessDone
from twisted.internet.defer import Deferred
from twisted.internet.protocol import ProcessProtocol

from twisted.protocols.basic import LineOnlyReceiver


RUN_ACTION = ActionType(
    action_type="flocker.common.runner:run",
    startFields=[
        Field.for_types(u"command", [list], u"The command.")
    ],
    successFields=[],
    description="Run a command.",
)
RUN_OUTPUT_MESSAGE = MessageType(
    message_type="flocker.common.runner:run:stdout",
    fields=[
        Field.for_types(u"line", [bytes], u"The output."),
    ],
    description=u"A line of command output.",
)
RUN_ERROR_MESSAGE = MessageType(
    message_type="flocker.common.runner:run:stderr",
    fields=[
        Field.for_types(u"line", [bytes], u"The error."),
    ],
    description=u"A line of command stderr.",
)

# Logging for the scp, upload, and download functions
SCP_ACTION = ActionType(

[CODE, RESPONSE, MESSAGE, DETAILS, REQUEST_ID, URL, METHOD],
)

LOCAL_IPS = Field(
    u"local_ips",
    repr,
    u"The IP addresses found on the target node."
)

API_IPS = Field(
    u"api_ips",
    repr,
    u"The IP addresses and instance_ids for all nodes."
)

COMPUTE_INSTANCE_ID_NOT_FOUND = MessageType(
    u"flocker:node:agents:blockdevice:openstack:compute_instance_id:not_found",
    [LOCAL_IPS, API_IPS],
    u"Unable to determine the instance ID of this node.",
)

CINDER_LOG_HEADER = u'flocker:node:agents:blockdevice:openstack'

# ActionType used by OpenStack storage driver.
OPENSTACK_ACTION = ActionType(
    CINDER_LOG_HEADER,
    [OPERATION],
    [],
    u"An IBlockDeviceAPI operation is executing using OpenStack"
    u"storage driver.")

CINDER_CREATE = u'flocker:node:agents:blockdevice:openstack:create_volume'

_LOG_SAVE = ActionType(u"flocker-control:persistence:save",
                       [_DEPLOYMENT_FIELD], [])

_UPGRADE_SOURCE_FIELD = Field.for_types(
    u"source_version", [int], u"Configuration version to upgrade from.")
_UPGRADE_TARGET_FIELD = Field.for_types(
    u"target_version", [int], u"Configuration version to upgrade to.")
_LOG_UPGRADE = ActionType(u"flocker-control:persistence:migrate_configuration",
                          [_DEPLOYMENT_FIELD, _UPGRADE_SOURCE_FIELD,
                           _UPGRADE_TARGET_FIELD, ], [])
_LOG_EXPIRE = MessageType(
    u"flocker-control:persistence:lease-expired",
    [Field(u"dataset_id", unicode), Field(u"node_id", unicode)],
    u"A lease for a dataset has expired.")

_LOG_UNCHANGED_DEPLOYMENT_NOT_SAVED = MessageType(
    u"flocker-control:persistence:unchanged-deployment-not-saved",
    [],
    u"The persistence service was told to save a deployment which is the same "
    u"as the already-saved deployment.  It has optimized this away."
)


class LeaseService(Service):
    """
    Manage leases.
    In particular, clear out expired leases once a second.

    :ivar _reactor: A ``IReactorTime`` provider.
    :ivar _persistence_service: The persistence service to act with.
    :ivar _lc: A ``twisted.internet.task.LoopingCall`` run every second
        to update the configured leases by releasing leases that have

from twisted.python.components import proxyForInterface
from twisted.python.filepath import FilePath
from twisted.internet.defer import succeed, fail
from twisted.internet.threads import deferToThread
from twisted.web.http import NOT_FOUND, INTERNAL_SERVER_ERROR

from ..common import (
    poll_until,
    retry_if, decorate_methods, with_retry, get_default_retry_steps,
)

from ..control._model import (
    RestartNever, RestartAlways, RestartOnFailure, pset_field, pvector_field)


LOG_CACHED_IMAGE = MessageType(
    u"flocker:node:docker:image_from_cache",
    [Field.for_types(u"image", [unicode], "The image ID.")],
    "An image was retrieved from the cache."
)


class AlreadyExists(Exception):
    """A unit with the given name already exists."""


@with_cmp(["address", "apierror"])
class AddressInUse(Exception):
    """
    The listen address for an exposed port was in use and could not be bound.
    """
    def __init__(self, address, apierror):

lambda path: path.path,
    u"The system device file for an attached block device."
)

CREATE_BLOCK_DEVICE_DATASET = ActionType(
    u"agent:blockdevice:create",
    [DATASET, MOUNTPOINT],
    [],
    u"A block-device-backed dataset is being created.",
)

# Really this is the successful completion of CREATE_BLOCK_DEVICE_DATASET.  It
# might be nice if these fields could just be added to the running action
# instead of being logged as a separate message (but still in the correct
# context).  Or maybe this is fine as-is.
BLOCK_DEVICE_DATASET_CREATED = MessageType(
    u"agent:blockdevice:created",
    [DEVICE_PATH, BLOCK_DEVICE_ID, DATASET_ID, BLOCK_DEVICE_SIZE,
     BLOCK_DEVICE_COMPUTE_INSTANCE_ID],
    u"A block-device-backed dataset has been created.",
)

DESTROY_BLOCK_DEVICE_DATASET = ActionType(
    u"agent:blockdevice:destroy",
    [DATASET_ID],
    [],
    u"A block-device-backed dataset is being destroyed.",
)

UNMOUNT_BLOCK_DEVICE = ActionType(
    u"agent:blockdevice:unmount",
    [DATASET_ID],

RUN_ACTION = ActionType(
    action_type="flocker.common.runner:run",
    startFields=[
        Field.for_types(u"command", [list], u"The command.")
    ],
    successFields=[],
    description="Run a command.",
)
RUN_OUTPUT_MESSAGE = MessageType(
    message_type="flocker.common.runner:run:stdout",
    fields=[
        Field.for_types(u"line", [bytes], u"The output."),
    ],
    description=u"A line of command output.",
)
RUN_ERROR_MESSAGE = MessageType(
    message_type="flocker.common.runner:run:stderr",
    fields=[
        Field.for_types(u"line", [bytes], u"The error."),
    ],
    description=u"A line of command stderr.",
)

# Logging for the scp, upload, and download functions
SCP_ACTION = ActionType(
    action_type="flocker.common.runner:scp",
    startFields=[
        Field.for_types(u"username", [bytes], u"SSH username."),
        Field.for_types(u"host", [bytes], u"SSH hostname."),
        Field(
            u"remote_path",
            lambda f: f.path,

CODE = Field.for_types("code", [int], u"The HTTP response code.")
MESSAGE = Field.for_types(
    "message", [bytes, unicode],
    u"A human-readable error message given by the response.",
)
DETAILS = Field.for_types("details", [dict], u"Extra details about the error.")
REQUEST_ID = Field.for_types(
    "request_id", [bytes, unicode],
    u"The unique identifier assigned by the server for this request.",
)
URL = Field.for_types("url", [bytes, unicode], u"The request URL.")
METHOD = Field.for_types("method", [bytes, unicode], u"The request method.")

NOVA_CLIENT_EXCEPTION = MessageType(
    u"openstack:nova_client_exception", [
        CODE,
        MESSAGE,
        DETAILS,
        REQUEST_ID,
        URL,
        METHOD,
    ],
)

RESPONSE = Field.for_types("response", [bytes, unicode], u"The response body.")

KEYSTONE_HTTP_ERROR = MessageType(
    u"openstack:keystone_http_error", [
        CODE,
        RESPONSE,

# Copyright ClusterHQ Inc.  See LICENSE file for details.

from eliot import Field, MessageType

CINDER_VOLUME = MessageType(
    u"flocker:functional:cinder:cinder_volume:created",
    [Field.for_types(
        u"id", [bytes, unicode],
        u"The Cinder-assigned unique identifier for the volume that was "
        u"created.",