systemtap/pcp integration pmda 0.01

systemtap/pcp integration pmda 0.01

[David Smith]

[-- Attachment #1: Type: text/plain, Size: 3439 bytes --]

Here's version 0.01 of my systemtap/pcp integration work that uses
systemtap (https://sourceware.org/systemtap/) to export JSON data and a
pcp (http://www.performancecopilot.org/) python pmda that reads and
processes the JSON data.

At this point, the systemtap side of things is *way* behind the pcp
side. The attached systemtap script has hand-crafted JSON schema and
data output functions. The script itself is based on some work that Will
Cohen has been doing to measure network latency. Note that the script
does output live data.

The pcp side of things in the attached python pmda is much farther along
than the systemtap side of th ings. It parses the JSON schema and data
that the systemtap script creates, and creates single-valued numeric or
string metrics for pcp.

Here's what the schema looks like that the systemtap script creates:

====
{
  "type": "object",
  "title": "root",
  "additionalProperties": false,
  "properties": {
    "generation": {
      "type": "integer",
      "additionalProperties": false
    },
    "data": {
      "type": "object",
      "additionalProperties": false,
      "properties": {
        "interface": {
          "type": "string",
          "description": "network xmit device",
          "minLength": 0,
          "additionalProperties": false
        },
        "xmit_count": {
          "type": "integer",
          "description": "number of packets for xmit device",
          "minimum": 0,
          "default": 0,
          "additionalProperties": false
        },
        "xmit_latency": {
          "type": "integer",
          "description": "sum of latency for xmit device",
          "minimum": 0,
          "default": 0,
          "additionalProperties": false
        }
      },
      "required": [
        "interface",
        "xmit_count",
        "xmit_latency"
      ]
    }
  }
}
====

Here's what the data produced by the systemtap script looks like:

====
{
  "generation": 1,
  "data": {
    "interface": "eth0",
    "xmit_count": 652046,
    "xmit_latency": 90
  }
}
====

Once the systemtap script is running and the pcp pmda is installed,
here's what the data produced in pcp looks like:

====
# pminfo -df stap_json

stap_json.root.xmit_latency
    Data Type: 64-bit int  InDom: PM_INDOM_NULL 0xffffffff
    Semantics: counter  Units: none
    value 2040

stap_json.root.xmit_count
    Data Type: 64-bit int  InDom: PM_INDOM_NULL 0xffffffff
    Semantics: counter  Units: none
    value 8856603

stap_json.root.interface
    Data Type: string  InDom: PM_INDOM_NULL 0xffffffff
    Semantics: instant  Units: none
    value "eth0"
====

So why is this version 0.01? Because there is a long way to go.

On the systemtap side of things, we don't really want to make a user
hand-craft his own JSON schema and data strings. So, a lot of work is
needed there.

On the pcp side of things, errors aren't handled well (the script will
just abort in several places), it only supports single-valued metrics,
it only supports a single systemtap script, etc. So, a lot of work is
needed there also.

But, I thought I'd go ahead and show this work in hopes of getting
feedback before I go too far down a wrong path.

(Note that I didn't attach some of the files needed by the pcp pmda to
actually get it installed, but I've checked them into the pcp.dsmith git
repository.)

-- 
David Smith
dsmith@redhat.com
Red Hat
http://www.redhat.com
256.217.0141 (direct)
256.837.0057 (fax)

[-- Attachment #2: pmdastap_json.python --]
[-- Type: text/plain, Size: 8441 bytes --]

#!/usr/bin/python
import json
import jsonschema
import collections
from pcp.pmda import PMDA, pmdaMetric, pmdaIndom, pmdaInstid
import cpmapi as c_api
from pcp.pmapi import pmUnits, pmContext as PCP

class STAP_JSON_PMDA(PMDA):
    def __init__(self, domain=255):
        # Load the schema and data.
        self.load_json_schema()
        self.load_json_data()

        # Make sure the data fits the schema.
        jsonschema.validate(self.json_data, self.schema)

        # Parse the schema header, looking for a name.
        self.pmda_name = ""
        self._parse_schema_header()
        PMDA.__init__(self, self.pmda_name, domain)

        self.configfile = "%s/%s/%s.conf" % (PCP.pmGetConfig('PCP_PMDAS_DIR'),
                                             self.pmda_name, self.pmda_name)
        self.connect_pmcd()

        # Parse the data portion of the schema header, creating
        # metrics as needed.
        self._parse_schema_data()

        self.set_fetch(self._fetch)
        self.set_fetch_callback(self._fetch_callback)
        self.set_store_callback(self._store_callback)
        self.set_user(PCP.pmGetConfig('PCP_USER'))

    def load_jscon_schema(self):
        # Load schema
        f = open("/proc/systemtap/json/schema")
        self.schema = json.load(f, object_pairs_hook=collections.OrderedDict)
        f.close()

    def load_json_data(self):
        # Load data
        f = open("/proc/systemtap/json/data")
        self.json_data = json.load(f, object_pairs_hook=collections.OrderedDict)
        f.close()

    def _parse_schema_header(self):
        '''
        Go through the schema, looking for information we can use to create
        the pcp representation of the schema. Note that we don't support
        every possible JSON schema, we're looking for certain items.

        Refer to the following link for details of JSON schemas:

        <http://tools.ietf.org/html/draft-zyp-json-schema-03>
        '''

        # First process the schema "header" information.
        self.data_header = None
        for (key, value) in self.schema.items():
            # 'type' (required): Just sanity check it.
            if key == "type":
                if not isinstance(value, unicode) or value != "object":
                    raise TypeError
            # 'title' (optional): Type check it.
            elif key == "title":
                if not isinstance(value, unicode):
                    raise TypeError
            # 'description' (optional): Type check it.
            elif key == "description":
                if not isinstance(value, unicode):
                    raise TypeError
            # 'additionalProperties' (optional): Ignore it.
            elif key == "additionalProperties":
                # Do nothing.
                pass
            # 'properties' (required): Type check it and save for later.
            elif key == "properties":
                if not isinstance(value, dict):
                    raise TypeError
                self.data_header = value
            # For everything else, raise an error.
            else:
                raise RuntimeError, "Unknown attribute '%s'" % key
        
        # Pick the right field for the PMDA name - prefer "title" over
        # "description".
        if self.schema.has_key("title"):
            self.pmda_name = self.schema["title"]
        elif self.schema.has_key("description"):
            self.pmda_name = self.schema["description"]
        else:
            raise RuntimeError, "No 'title' or 'description' field in schema header"
    
    def _parse_schema_data(self):
        # If we're here, we know the "header" was reasonable. Now process
        # "properties", which is the data "header".
        if not self.data_header:
            raise RuntimeError, "Schema has no 'properties' attribute"
        data_properties = None
        for (key, value) in self.data_header.items():
            # 'generation' (required): Just sanity check it.
            if key == "generation":
                if not isinstance(value, dict):
                    raise TypeError
            # 'data' (required): Type check it.
            elif key == "data":
                if not isinstance(value, dict) \
                   or not value.has_key("properties") \
                   or not isinstance(value["properties"], dict):
                    raise TypeError
                data_properties = value["properties"]
            # For everything else, raise an error.
            else:
                raise RuntimeError, "Unknown attribute '%s'" % key
        
        # If we're here, we know the data "header" was reasonable. Now process
        # "properties.data.properties", which is the real data description.
        if not data_properties:
            raise RuntimeError, "Schema has no 'properties.data.properties' attribute"
        metric_idx = 0
        for (name, attributes) in data_properties.items():
            metric_info = {}
            metric_info["name"] = name
            metric_info["oneline"] = ""
            for (key, value) in attributes.items():
                # 'type' (required): Sanity check it and save it.
                if key == "type":
                    if not isinstance(value, unicode):
                        raise TypeError
                    if value == "string":
                        metric_info['type'] = c_api.PM_TYPE_STRING
                    elif value == "integer":
                        metric_info['type'] = c_api.PM_TYPE_64
                # 'description' (optional): Type check it and save it.
        	elif key == "description":
                    if not isinstance(value, unicode):
                        raise TypeError
                    metric_info['oneline'] = value
                # 'minLength' (optional): Ignore it (for now).
            	elif key == "minLength":
                    # Do nothing.
                    pass
                # 'additionalProperties' (optional): Ignore it.
            	elif key == "additionalProperties":
                    # Do nothing.
                    pass
                # 'minimum' (optional): Ignore it (for now).
            	elif key == "minimum":
                    # Do nothing for now.
                    pass
                # 'default' (optional): Ignore it (for now).
            	elif key == "default":
                    # Do nothing for now.
                    pass
                # For everything else, raise an error.
            	else:
                    raise RuntimeError, \
                        ("Schema for '%s' has an unknown attribute '%s'"
                         % (name, key)) 

            # Make sure we have everything we need.
            if not metric_info.has_key("type"):
                raise RuntimeError, ("Schema for '%s' has no 'type' attribute"
                                     % name)

            # Add the metric.
            metric_info["pmid"] = self.pmid(0, metric_idx)
            metric = pmdaMetric(metric_info["pmid"], metric_info["type"],
                                c_api.PM_INDOM_NULL, c_api.PM_SEM_COUNTER,
                                pmUnits(0, 0, 0, 0, 0, 0))

            self.add_metric('stap_json.' + self.pmda_name + '.' + name, metric,
                            metric_info["oneline"])
            self.data_idx[metric_idx] = name
            metric_idx += 1

    def _fetch(self):
        ''' Called once per "fetch" PDU, before callbacks '''
        self.load_json_data()

    def _fetch_callback(self, cluster, item, inst):
        '''
        Main fetch callback. Returns a list of value,status (single
        pair) for requested pmid/inst.
        '''
        if cluser != 0 or inst != 0:
            return [c_api.PM_ERR_PMID, 0]
        try:
            # Note that this will have to get more complicated once we
            # have instances. This only handles single valued items.
            metric_name = self.data_idx[item]
            return [metric_name, self.json_data['data'][metric_name]]
        except:
            return [c_api.PM_ERR_PMID, 0]

    def _store_callback(self, cluster, item, inst, val):
        '''
        Store callback, executed when a request to write to a metric
        happens. Returns a single value.
        '''
        # Since we don't support storing values, always fail.
        return c_api.PM_ERR_PERMISSION


if __name__ == '__main__':
    STAP_JSON_PMDA(255).run()

[-- Attachment #3: net_xmit_json3.stp --]
[-- Type: text/plain, Size: 3513 bytes --]

// This script tracks time between packet queue and xmit.
// The information is provided to userspace using Memory Mapped Values (mmv)
// which are updated every second and are readable by a PCP pmda.

global net_device

probe procfs("schema").read.maxsize(8192)
{
  # Note: This is the "pretty-printed" version of the schema, intended
  # to be read by humans. We could remove the whitespace and newlines
  # if we wanted to make the output shorter (but less readable).
  #
  # Note 2: Note that we have to break this long string into more than
  # 1 assignment since we're bumping up against MAXSTRINGLEN. Procfs
  # $value can hold more than MAXSTRINGLEN because of the
  # '.maxsize(N)' parameter above.
  $value =
    "{\n"
    "  \"type\": \"object\",\n"
    "  \"title\": \"root\",\n"
    "  \"additionalProperties\": false,\n"
    "  \"properties\": {\n"
    "    \"generation\": {\n"
    "      \"type\": \"integer\",\n"
    "      \"additionalProperties\": false\n"
    "    },\n"
  $value .=
    "    \"data\": {\n"
    "      \"type\": \"object\",\n"
    "      \"additionalProperties\": false,\n"
    "      \"properties\": {\n"
    "        \"interface\": {\n"
    "          \"type\": \"string\",\n"
    "          \"description\": \"network xmit device\",\n"
    "          \"minLength\": 0,\n"
    "          \"additionalProperties\": false\n"
    "        },\n"
  $value .=
    "        \"xmit_count\": {\n"
    "          \"type\": \"integer\",\n"
    "          \"description\": \"number of packets for xmit device\",\n"
    "          \"minimum\": 0,\n"
    "          \"default\": 0,\n"
    "          \"additionalProperties\": false\n"
    "        },\n"
    "        \"xmit_latency\": {\n"
    "          \"type\": \"integer\",\n"
    "          \"description\": \"sum of latency for xmit device\",\n"
    "          \"minimum\": 0,\n"
    "          \"default\": 0,\n"
    "          \"additionalProperties\": false\n"
    "        }\n"
    "      },\n"
  $value .=
    "      \"required\": [\n"
    "        \"interface\",\n"
    "        \"xmit_count\",\n"
    "        \"xmit_latency\"\n"
    "      ]\n"
    "    }\n"
    "  }\n"
    "}\n"
}

probe procfs("data").read.maxsize(8192)
{
  # NOTE: This is the "pretty-printed" version of the data, intended
  # to be read by humans. We could remove the whitespace and newlines
  # if we wanted to make the output shorter (but less readable).

  $value =
    "{\n"
    "  \"generation\": 1,\n"
    "  \"data\": {\n"
  if (@count(skb_queue_t[net_device])) {
    $value .= sprintf("    \"interface\": \"%s\",\n    \"xmit_count\": %d,\n    \"xmit_latency\": %d\n",
			net_device,
			@sum(skb_queue_t[net_device]),
			@count(skb_queue_t[net_device]))
  } else {
    $value .= sprintf("    \"interface\": \"%s\",\n    \"xmit_count\": %d,\n    \"xmit_latency\": %d\n",
			net_device, 0, 0)
  }
  $value .=
    "  }\n"
    "}\n"
}

// Set up the mmv and make it available
probe begin
{
  // fallback instance device "eth0" if none specified
  if (argc == 0) {
    argv[1] = "eth0"
  }
  
  // remember the network device
  net_device = argv[1]
}

// probes to track the information

global skb_queue_start_t, skb_queue_t

probe kernel.trace("net_dev_queue") {
  skb_queue_start_t[$skb] = gettimeofday_ns();
}

probe kernel.trace("net_dev_start_xmit"), kernel.trace("net_dev_xmit") {
  t = gettimeofday_ns();
  st = skb_queue_start_t[$skb]
  if (st){
    skb_queue_t[kernel_string($dev->name)] <<< t - st
    delete skb_queue_start_t[$skb]
  }
}