<commit_before># Copyright (c) 2016 VMware, Inc. All rights reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.
#

import json
import six
import uuid

import eventlet
eventlet.monkey_patch()  # for using oslo.messaging w/ eventlet executor

from futurist import periodics
from oslo_concurrency import lockutils
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_log import log as logging
import oslo_messaging as messaging
from oslo_messaging import exceptions as messaging_exceptions
from oslo_utils import importutils
from oslo_utils import strutils
from oslo_utils import uuidutils

from congress.datasources import constants
from congress.db import datasources as datasources_db
from congress.dse2 import control_bus
from congress import exception


LOG = logging.getLogger(__name__)


_dse_opts = [
    cfg.StrOpt('bus_id', default='bus',
               help='Unique ID of this DSE bus'),
    cfg.IntOpt('ping_timeout', default=5,
               help='RPC short timeout in seconds; used to ping destination'),
    cfg.IntOpt('long_timeout', default=120,
               help='RPC long timeout in seconds; used on potentially long '
                    'running requests such as datasource action and PE row '
                    'query'),
    cfg.IntOpt('time_to_resub', default=10,
               help='Time in seconds which a subscriber will wait for missing '
                    'update before attempting to resubscribe from publisher'),
    cfg.BoolOpt('execute_action_retry', default=False,
                help='Set the flag to True to make Congress retry execute '
                     'actions; may cause duplicate executions.'),
    cfg.IntOpt('execute_action_retry_timeout', default=600,
               help='The number of seconds to retry execute action before '
                    'giving up. Zero or negative value means never give up.'),
]
cfg.CONF.register_opts(_dse_opts, group='dse')


class DseNode(object):
    """Addressable entity participating on the DSE message bus.

    The Data Services Engine (DSE) is comprised of one or more DseNode
    instances that each may run one or more DataService instances.  All
    communication between data services uses the DseNode interface.

    Attributes:
        node_id: The unique ID of this node on the DSE.
        messaging_config: Configuration options for the message bus.  See
                          oslo.messaging for more details.
        node_rpc_endpoints: List of object instances exposing a remotely
                            invokable interface.
    """
    RPC_VERSION = '1.0'
    EXCHANGE = 'congress'
    CONTROL_TOPIC = 'congress-control'
    SERVICE_TOPIC_PREFIX = 'congress-service-'

    def node_rpc_target(self, namespace=None, server=None, fanout=False):
        return messaging.Target(exchange=self.EXCHANGE,
                                topic=self._add_partition(self.CONTROL_TOPIC),
                                version=self.RPC_VERSION,
                                namespace=namespace,
                                server=server,
                                fanout=fanout)

    def service_rpc_target(self, service_id, namespace=None, server=None,
                           fanout=False):
        topic = self._add_partition(self.SERVICE_TOPIC_PREFIX + service_id)
        return messaging.Target(exchange=self.EXCHANGE,
                                topic=topic,
                                version=self.RPC_VERSION,
                                namespace=namespace,
                                server=server,
                                fanout=fanout)

    def _add_partition(self, topic, partition_id=None):
        """Create a seed-specific version of an oslo-messaging topic."""
        partition_id = partition_id or self.partition_id
        if partition_id is None:
            return topic
        return topic + "-" + str(partition_id)

    def __init__(self, messaging_config, node_id, node_rpc_endpoints,
                 partition_id=None):
        # Note(ekcs): temporary setting to disable use of diffs and sequencing
        #   to avoid muddying the process of a first dse2 system test.
        # TODO(ekcs,dse2): remove when differential update is standard
        self.always_snapshot = False
        self.messaging_config = messaging_config
        self.node_id = node_id
        self.node_rpc_endpoints = node_rpc_endpoints
        # unique identifier shared by all nodes that can communicate
        self.partition_id = partition_id or cfg.CONF.dse.bus_id or "bus"
        self.node_rpc_endpoints.append(DseNodeEndpoints(self))
        self._running = False
        self._services = []
        self.instance = uuid.uuid4()  # uuid to help recognize node_id clash
        # TODO(dse2): add detection and logging/rectifying for node_id clash?
        self.context = self._message_context()
        self.transport = messaging.get_transport(
            self.messaging_config,
            allowed_remote_exmods=[exception.__name__, ])
        self._rpctarget = self.node_rpc_target(self.node_id, self.node_id)
        self._rpc_server = messaging.get_rpc_server(
            self.transport, self._rpctarget, self.node_rpc_endpoints,
            executor='eventlet')

        # # keep track of what publisher/tables local services subscribe to
        # subscribers indexed by publisher and table:
        # {publisher_id ->
        #     {table_name -> set_of_subscriber_ids}}
        self.subscriptions = {}

        # Note(ekcs): A little strange that _control_bus starts before self?
        self._control_bus = control_bus.DseNodeControlBus(self)
        self.register_service(self._control_bus)
        # load configured drivers
        self.loaded_drivers = self.load_drivers()
        self.periodic_tasks = None
        self.sync_thread = None
        self.start()

    def __del__(self):
        self.stop()
        self.wait()

    def __repr__(self):
        return self.__class__.__name__ + "<%s>" % self.node_id

    def _message_context(self):
        return {'node_id': self.node_id, 'instance': str(self.instance)}

    # Note(thread-safety): blocking function
    def register_service(self, service):
        assert service.node is None
        if self.service_object(service.service_id):
            msg = ('Service %s already exsists on the node %s'
                   % (service.service_id, self.node_id))
            raise exception.DataServiceError(msg)

        service.always_snapshot = self.always_snapshot
        service.node = self
        self._services.append(service)
        service._target = self.service_rpc_target(service.service_id,
                                                  server=self.node_id)
        service._rpc_server = messaging.get_rpc_server(
            self.transport, service._target, service.rpc_endpoints(),
            executor='eventlet')

        service.start()

        LOG.debug('<%s> Service %s RPC Server listening on %s',
                  self.node_id, service.service_id, service._target)

    # Note(thread-safety): blocking function
    def unregister_service(self, service_id=None, uuid_=None):
        """Unregister service from DseNode matching on service_id or uuid_

        Only one should be supplied. No-op if no matching service found.
        """
        LOG.debug("unregistering service %s on node %s", service_id,
                  self.node_id)
        service = self.service_object(service_id=service_id, uuid_=uuid_)
        if service is not None:
            self._services.remove(service)
            service.stop()
            # Note(thread-safety): blocking call
            service.wait()

    def get_services(self, hidden=False):
        """Return all local service objects."""
        if hidden:
            return self._services
        return [s for s in self._services if s.service_id[0] != '_']

    def get_global_service_names(self, hidden=False):
        """Return names of all services on all nodes."""
        services = self.get_services(hidden=hidden)
        local_services = [s.service_id for s in services]
        # Also, check services registered on other nodes
        peer_nodes = self.dse_status()['peers']
        peer_services = []
        for node in peer_nodes.values():
            peer_services.extend(
                [srv['service_id'] for srv in node['services']])
        return set(local_services + peer_services)

    def service_object(self, service_id=None, uuid_=None):
        """Return the service object requested.

        Search by service_id or uuid_ (only one should be supplied).
        None if not found.
        """
        if service_id is not None:
            if uuid_ is not None:
                raise TypeError('service_object() cannot accept both args '
                                'service_id and uuid_')
            for s in self._services:
                if s.service_id == service_id:
                    return s
        elif uuid_ is not None:
            for s in self._services:
                if getattr(s, 'ds_id', None) == uuid_:
                    return s
        else:
            raise TypeError('service_object() requires service_id or '
                            'uuid_ argument, but neither is given.')
        return None

    def start(self):
        LOG.debug("<%s> DSE Node '%s' starting with %s sevices...",
                  self.node_id, self.node_id, len(self._services))

        # Start Node RPC server
        self._rpc_server.start()
        LOG.debug('<%s> Node RPC Server listening on %s',
                  self.node_id, self._rpctarget)

        # Start Service RPC server(s)
        for s in self._services:
            s.start()
            LOG.debug('<%s> Service %s RPC Server listening on %s',
                      self.node_id, s.service_id, s._target)

        self._running = True

    def stop(self):
        if self._running is False:
            return

        LOG.info("Stopping DSE node '%s'", self.node_id)
        self.stop_datasource_synchronizer()
        for s in self._services:
            s.stop()
        self._rpc_server.stop()
        self._running = False

    # Note(thread-safety): blocking function
    def wait(self):
        for s in self._services:
            # Note(thread-safety): blocking call
            s.wait()
        # Note(thread-safety): blocking call
        self._rpc_server.wait()

    def dse_status(self):
        """Return latest observation of DSE status."""
        return self._control_bus.dse_status()

    def is_valid_service(self, service_id):
        return service_id in self.get_global_service_names(hidden=True)

    # Note(thread-safety): blocking function
    def invoke_node_rpc(self, node_id, method, kwargs=None, timeout=None):
        """Invoke RPC method on a DSE Node.

        Args:
            node_id: The ID of the node on which to invoke the call.
            method: The method name to call.
            kwargs: A dict of method arguments.

        Returns:
            The result of the method invocation.

        Raises: MessagingTimeout, RemoteError, MessageDeliveryFailure
        """
        if kwargs is None:
            kwargs = {}
        target = self.node_rpc_target(server=node_id)
        LOG.trace("<%s> Invoking RPC '%s' on %s", self.node_id, method, target)
        client = messaging.RPCClient(self.transport, target, timeout=timeout)
        return client.call(self.context, method, **kwargs)

    # Note(thread-safety): blocking function
    def broadcast_node_rpc(self, method, kwargs=None):
        """Invoke RPC method on all DSE Nodes.

        Args:
            method: The method name to call.
            kwargs: A dict of method arguments.

        Returns:
            None - Methods are invoked asynchronously and results are dropped.

        Raises: RemoteError, MessageDeliveryFailure
        """
        if kwargs is None:
            kwargs = {}
        target = self.node_rpc_target(fanout=True)
        LOG.trace("<%s> Casting RPC '%s' on %s", self.node_id, method, target)
        client = messaging.RPCClient(self.transport, target)
        client.cast(self.context, method, **kwargs)

    # Note(thread-safety): blocking function
    def invoke_service_rpc(
            self, service_id, method, kwargs=None, timeout=None, local=False,
            retry=None):
        """Invoke RPC method on a DSE Service.

        Args:
            service_id: The ID of the data service on which to invoke the call.
            method: The method name to call.
            kwargs: A dict of method arguments.

        Returns:
            The result of the method invocation.

        Raises: MessagingTimeout, RemoteError, MessageDeliveryFailure, NotFound
        """
        target = self.service_rpc_target(
            service_id, server=(self.node_id if local else None))
        LOG.trace("<%s> Preparing to invoking RPC '%s' on %s",
                  self.node_id, method, target)
        client = messaging.RPCClient(self.transport, target, timeout=timeout,
                                     retry=retry)
        if not self.is_valid_service(service_id):
            try:
                # First ping the destination to fail fast if unresponsive
                LOG.trace("<%s> Checking responsiveness before invoking RPC "
                          "'%s' on %s", self.node_id, method, target)
                client.prepare(timeout=cfg.CONF.dse.ping_timeout).call(
                    self.context, 'ping')
            except (messaging_exceptions.MessagingTimeout,
                    messaging_exceptions.MessageDeliveryFailure):
                msg = "service '%s' could not be found"
                raise exception.NotFound(msg % service_id)
        if kwargs is None:
            kwargs = {}
        try:
            LOG.trace(
                "<%s> Invoking RPC '%s' on %s", self.node_id, method, target)
            result = client.call(self.context, method, **kwargs)
        except (messaging_exceptions.MessagingTimeout,
                messaging_exceptions.MessageDeliveryFailure):
            msg = "Request to service '%s' timed out"
            raise exception.NotFound(msg % service_id)
        LOG.trace("<%s> RPC call returned: %s", self.node_id, result)
        return result<commit_msg>Raise BadRequest while pushing data to non-push ds<commit_after># Copyright (c) 2016 VMware, Inc. All rights reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.
#

import json
import six
import uuid

import eventlet
eventlet.monkey_patch()  # for using oslo.messaging w/ eventlet executor

from futurist import periodics
from oslo_concurrency import lockutils
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_log import log as logging
import oslo_messaging as messaging
from oslo_messaging import exceptions as messaging_exceptions
from oslo_messaging.rpc import dispatcher
from oslo_utils import importutils
from oslo_utils import strutils
from oslo_utils import uuidutils

from congress.datasources import constants
from congress.db import datasources as datasources_db
from congress.dse2 import control_bus
from congress import exception


LOG = logging.getLogger(__name__)


_dse_opts = [
    cfg.StrOpt('bus_id', default='bus',
               help='Unique ID of this DSE bus'),
    cfg.IntOpt('ping_timeout', default=5,
               help='RPC short timeout in seconds; used to ping destination'),
    cfg.IntOpt('long_timeout', default=120,
               help='RPC long timeout in seconds; used on potentially long '
                    'running requests such as datasource action and PE row '
                    'query'),
    cfg.IntOpt('time_to_resub', default=10,
               help='Time in seconds which a subscriber will wait for missing '
                    'update before attempting to resubscribe from publisher'),
    cfg.BoolOpt('execute_action_retry', default=False,
                help='Set the flag to True to make Congress retry execute '
                     'actions; may cause duplicate executions.'),
    cfg.IntOpt('execute_action_retry_timeout', default=600,
               help='The number of seconds to retry execute action before '
                    'giving up. Zero or negative value means never give up.'),
]
cfg.CONF.register_opts(_dse_opts, group='dse')


class DseNode(object):
    """Addressable entity participating on the DSE message bus.

    The Data Services Engine (DSE) is comprised of one or more DseNode
    instances that each may run one or more DataService instances.  All
    communication between data services uses the DseNode interface.

    Attributes:
        node_id: The unique ID of this node on the DSE.
        messaging_config: Configuration options for the message bus.  See
                          oslo.messaging for more details.
        node_rpc_endpoints: List of object instances exposing a remotely
                            invokable interface.
    """
    RPC_VERSION = '1.0'
    EXCHANGE = 'congress'
    CONTROL_TOPIC = 'congress-control'
    SERVICE_TOPIC_PREFIX = 'congress-service-'

    def node_rpc_target(self, namespace=None, server=None, fanout=False):
        return messaging.Target(exchange=self.EXCHANGE,
                                topic=self._add_partition(self.CONTROL_TOPIC),
                                version=self.RPC_VERSION,
                                namespace=namespace,
                                server=server,
                                fanout=fanout)

    def service_rpc_target(self, service_id, namespace=None, server=None,
                           fanout=False):
        topic = self._add_partition(self.SERVICE_TOPIC_PREFIX + service_id)
        return messaging.Target(exchange=self.EXCHANGE,
                                topic=topic,
                                version=self.RPC_VERSION,
                                namespace=namespace,
                                server=server,
                                fanout=fanout)

    def _add_partition(self, topic, partition_id=None):
        """Create a seed-specific version of an oslo-messaging topic."""
        partition_id = partition_id or self.partition_id
        if partition_id is None:
            return topic
        return topic + "-" + str(partition_id)

    def __init__(self, messaging_config, node_id, node_rpc_endpoints,
                 partition_id=None):
        # Note(ekcs): temporary setting to disable use of diffs and sequencing
        #   to avoid muddying the process of a first dse2 system test.
        # TODO(ekcs,dse2): remove when differential update is standard
        self.always_snapshot = False
        self.messaging_config = messaging_config
        self.node_id = node_id
        self.node_rpc_endpoints = node_rpc_endpoints
        # unique identifier shared by all nodes that can communicate
        self.partition_id = partition_id or cfg.CONF.dse.bus_id or "bus"
        self.node_rpc_endpoints.append(DseNodeEndpoints(self))
        self._running = False
        self._services = []
        self.instance = uuid.uuid4()  # uuid to help recognize node_id clash
        # TODO(dse2): add detection and logging/rectifying for node_id clash?
        self.context = self._message_context()
        self.transport = messaging.get_transport(
            self.messaging_config,
            allowed_remote_exmods=[exception.__name__, dispatcher.__name__, ])
        self._rpctarget = self.node_rpc_target(self.node_id, self.node_id)
        self._rpc_server = messaging.get_rpc_server(
            self.transport, self._rpctarget, self.node_rpc_endpoints,
            executor='eventlet')

        # # keep track of what publisher/tables local services subscribe to
        # subscribers indexed by publisher and table:
        # {publisher_id ->
        #     {table_name -> set_of_subscriber_ids}}
        self.subscriptions = {}

        # Note(ekcs): A little strange that _control_bus starts before self?
        self._control_bus = control_bus.DseNodeControlBus(self)
        self.register_service(self._control_bus)
        # load configured drivers
        self.loaded_drivers = self.load_drivers()
        self.periodic_tasks = None
        self.sync_thread = None
        self.start()

    def __del__(self):
        self.stop()
        self.wait()

    def __repr__(self):
        return self.__class__.__name__ + "<%s>" % self.node_id

    def _message_context(self):
        return {'node_id': self.node_id, 'instance': str(self.instance)}

    # Note(thread-safety): blocking function
    def register_service(self, service):
        assert service.node is None
        if self.service_object(service.service_id):
            msg = ('Service %s already exsists on the node %s'
                   % (service.service_id, self.node_id))
            raise exception.DataServiceError(msg)

        service.always_snapshot = self.always_snapshot
        service.node = self
        self._services.append(service)
        service._target = self.service_rpc_target(service.service_id,
                                                  server=self.node_id)
        service._rpc_server = messaging.get_rpc_server(
            self.transport, service._target, service.rpc_endpoints(),
            executor='eventlet')

        service.start()

        LOG.debug('<%s> Service %s RPC Server listening on %s',
                  self.node_id, service.service_id, service._target)

    # Note(thread-safety): blocking function
    def unregister_service(self, service_id=None, uuid_=None):
        """Unregister service from DseNode matching on service_id or uuid_

        Only one should be supplied. No-op if no matching service found.
        """
        LOG.debug("unregistering service %s on node %s", service_id,
                  self.node_id)
        service = self.service_object(service_id=service_id, uuid_=uuid_)
        if service is not None:
            self._services.remove(service)
            service.stop()
            # Note(thread-safety): blocking call
            service.wait()

    def get_services(self, hidden=False):
        """Return all local service objects."""
        if hidden:
            return self._services
        return [s for s in self._services if s.service_id[0] != '_']

    def get_global_service_names(self, hidden=False):
        """Return names of all services on all nodes."""
        services = self.get_services(hidden=hidden)
        local_services = [s.service_id for s in services]
        # Also, check services registered on other nodes
        peer_nodes = self.dse_status()['peers']
        peer_services = []
        for node in peer_nodes.values():
            peer_services.extend(
                [srv['service_id'] for srv in node['services']])
        return set(local_services + peer_services)

    def service_object(self, service_id=None, uuid_=None):
        """Return the service object requested.

        Search by service_id or uuid_ (only one should be supplied).
        None if not found.
        """
        if service_id is not None:
            if uuid_ is not None:
                raise TypeError('service_object() cannot accept both args '
                                'service_id and uuid_')
            for s in self._services:
                if s.service_id == service_id:
                    return s
        elif uuid_ is not None:
            for s in self._services:
                if getattr(s, 'ds_id', None) == uuid_:
                    return s
        else:
            raise TypeError('service_object() requires service_id or '
                            'uuid_ argument, but neither is given.')
        return None

    def start(self):
        LOG.debug("<%s> DSE Node '%s' starting with %s sevices...",
                  self.node_id, self.node_id, len(self._services))

        # Start Node RPC server
        self._rpc_server.start()
        LOG.debug('<%s> Node RPC Server listening on %s',
                  self.node_id, self._rpctarget)

        # Start Service RPC server(s)
        for s in self._services:
            s.start()
            LOG.debug('<%s> Service %s RPC Server listening on %s',
                      self.node_id, s.service_id, s._target)

        self._running = True

    def stop(self):
        if self._running is False:
            return

        LOG.info("Stopping DSE node '%s'", self.node_id)
        self.stop_datasource_synchronizer()
        for s in self._services:
            s.stop()
        self._rpc_server.stop()
        self._running = False

    # Note(thread-safety): blocking function
    def wait(self):
        for s in self._services:
            # Note(thread-safety): blocking call
            s.wait()
        # Note(thread-safety): blocking call
        self._rpc_server.wait()

    def dse_status(self):
        """Return latest observation of DSE status."""
        return self._control_bus.dse_status()

    def is_valid_service(self, service_id):
        return service_id in self.get_global_service_names(hidden=True)

    # Note(thread-safety): blocking function
    def invoke_node_rpc(self, node_id, method, kwargs=None, timeout=None):
        """Invoke RPC method on a DSE Node.

        Args:
            node_id: The ID of the node on which to invoke the call.
            method: The method name to call.
            kwargs: A dict of method arguments.

        Returns:
            The result of the method invocation.

        Raises: MessagingTimeout, RemoteError, MessageDeliveryFailure
        """
        if kwargs is None:
            kwargs = {}
        target = self.node_rpc_target(server=node_id)
        LOG.trace("<%s> Invoking RPC '%s' on %s", self.node_id, method, target)
        client = messaging.RPCClient(self.transport, target, timeout=timeout)
        return client.call(self.context, method, **kwargs)

    # Note(thread-safety): blocking function
    def broadcast_node_rpc(self, method, kwargs=None):
        """Invoke RPC method on all DSE Nodes.

        Args:
            method: The method name to call.
            kwargs: A dict of method arguments.

        Returns:
            None - Methods are invoked asynchronously and results are dropped.

        Raises: RemoteError, MessageDeliveryFailure
        """
        if kwargs is None:
            kwargs = {}
        target = self.node_rpc_target(fanout=True)
        LOG.trace("<%s> Casting RPC '%s' on %s", self.node_id, method, target)
        client = messaging.RPCClient(self.transport, target)
        client.cast(self.context, method, **kwargs)

    # Note(thread-safety): blocking function
    def invoke_service_rpc(
            self, service_id, method, kwargs=None, timeout=None, local=False,
            retry=None):
        """Invoke RPC method on a DSE Service.

        Args:
            service_id: The ID of the data service on which to invoke the call.
            method: The method name to call.
            kwargs: A dict of method arguments.

        Returns:
            The result of the method invocation.

        Raises: MessagingTimeout, RemoteError, MessageDeliveryFailure, NotFound
        """
        target = self.service_rpc_target(
            service_id, server=(self.node_id if local else None))
        LOG.trace("<%s> Preparing to invoking RPC '%s' on %s",
                  self.node_id, method, target)
        client = messaging.RPCClient(self.transport, target, timeout=timeout,
                                     retry=retry)
        if not self.is_valid_service(service_id):
            try:
                # First ping the destination to fail fast if unresponsive
                LOG.trace("<%s> Checking responsiveness before invoking RPC "
                          "'%s' on %s", self.node_id, method, target)
                client.prepare(timeout=cfg.CONF.dse.ping_timeout).call(
                    self.context, 'ping')
            except (messaging_exceptions.MessagingTimeout,
                    messaging_exceptions.MessageDeliveryFailure):
                msg = "service '%s' could not be found"
                raise exception.NotFound(msg % service_id)
        if kwargs is None:
            kwargs = {}
        try:
            LOG.trace(
                "<%s> Invoking RPC '%s' on %s", self.node_id, method, target)
            result = client.call(self.context, method, **kwargs)
        except dispatcher.NoSuchMethod:
            msg = "Method %s not supported for datasource %s"
            LOG.exception(msg, method, service_id)
            raise exception.BadRequest(msg % (method, service_id))
        except (messaging_exceptions.MessagingTimeout,
                messaging_exceptions.MessageDeliveryFailure):
            msg = "Request to service '%s' timed out"
            raise exception.NotFound(msg % service_id)
        LOG.trace("<%s> RPC call returned: %s", self.node_id, result)
        return result
{'fraction_non_alphanumeric': 0.08183759258023621, 'fraction_numerical': 0.0016681123640488423, 'mean_word_length': 2.190187313750532, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2727368', 'n_tokens_mistral': 8140, 'n_tokens_neox': 7877, 'n_words': 2434}