Complete refactor in order to support RC-32

The influxdb RC-32 is quite strange, uses a UDP/IP line protocol with
batch updates.

This commit will cover the base UDP/IP protocol as first implementation

.travis.yml

@@ -1,16 +1,15 @@
 language: php
 
 php:
-  - 5.4
   - 5.5
   - 5.6
 
 before_install:
   - sudo apt-get update
   - sudo apt-get install libcurl4-openssl-dev libjson0-dev
-  - wget http://get.influxdb.org/influxdb_0.9.0-rc30_amd64.deb 
+  - wget http://get.influxdb.org/influxdb_0.9.0-rc32_amd64.deb 
   - sudo useradd influxdb
-  - sudo dpkg -i influxdb_0.9.0-rc30_amd64.deb
+  - sudo dpkg -i influxdb_0.9.0-rc32_amd64.deb
   - sudo cp ./scripts/influxdb_conf.toml /etc/opt/influxdb/influxdb.conf
   - travis_retry sudo service influxdb restart
   - sudo service influxdb status

scripts/influxdb_conf.toml

@@ -1,118 +1,74 @@
-# Welcome to the InfluxDB configuration file.
+[meta]
+  dir = "/var/opt/influxdb/meta"
+  hostname = "localhost"
+  bind-address = ":8088"
+  retention-autocreate = true
+  election-timeout = "1s"
+  heartbeat-timeout = "1s"
+  leader-lease-timeout = "500ms"
+  commit-timeout = "50ms"
 
-# If hostname (on the OS) doesn't return a name that can be resolved by the other
+[data]
-# systems in the cluster, you'll have to set the hostname to an IP or something
+  dir = "/var/opt/influxdb/data"
-# that can be resolved here.
+  retention-auto-create = true
-# hostname = ""
+  retention-check-enabled = true
-bind-address = "0.0.0.0"
+  retention-check-period = "10m0s"
+  retention-create-period = "45m0s"
 
-# The default cluster and API port
+[cluster]
-port = 8086
+  shard-writer-timeout = "5s"
 
-# Once every 24 hours InfluxDB will report anonymous data to m.influxdb.com
+[retention]
-# The data includes raft id (random 8 bytes), os, arch and version
+  enabled = true
-# We don't track ip addresses of servers reporting. This is only used
+  check-interval = "10m0s"
-# to track the number of instances running and the versions, which
-# is very helpful for us.
-# Change this option to true to disable reporting.
-reporting-disabled = false
 
-# Controls settings for initial start-up. Once a node is successfully started,
-# these settings are ignored.  If a node is started with the -join flag,
-# these settings are ignored.
-[initialization]
-join-urls = "" # Comma-delimited URLs, in the form http://host:port, for joining another cluster.
-
-# Control authentication
-# If not set authetication is DISABLED. Be sure to explicitly set this flag to
-# true if you want authentication.
-[authentication]
-enabled = false
-
-# Configure the admin server
 [admin]
   enabled = true
-port = 8083
+  bind-address = ":8083"
 
-# Configure the HTTP API endpoint. All time-series data and queries uses this endpoint.
+[http]
-[api]
+  enabled = true
-# ssl-port = 8087    # SSL support is enabled if you set a port and cert
+  bind-address = ":8086"
-# ssl-cert = "/path/to/cert.pem"
+  auth-enabled = false
+  log-enabled = true
+  write-tracing = false
+  pprof-enabled = false
 
-# Configure the Graphite plugins.
-[[graphite]] # 1 or more of these sections may be present.
-enabled = false
-# protocol = "" # Set to "tcp" or "udp"
-# address = "0.0.0.0" # If not set, is actually set to bind-address.
-# port = 2003
-# name-position = "last"
-# name-separator = "-"
-# database = ""  # store graphite data in this database
-
-# Configure the collectd input.
 [collectd]
   enabled = false
-#address = "0.0.0.0" # If not set, is actually set to bind-address.
+  bind-address = ""
-#port = 25827
+  database = ""
-#database = "collectd_database"
+  typesdb = ""
-#typesdb = "types.db"
 
-# Configure the OpenTSDB input.
 [opentsdb]
   enabled = false
-#address = "0.0.0.0" # If not set, is actually set to bind-address.
+  bind-address = ""
-#port = 4242
+  database = ""
-#database = "opentsdb_database"
+  retention-policy = ""
 
-# Configure UDP listener for series data.
 [udp]
   enabled = true
-bind-address = "0.0.0.0"
+  bind-address = ":4444"
-port = 4444
+  database = "udp.test"
+  batch-size = 0
+  batch-timeout = "10ns"
 
-# Broker configuration. Brokers are nodes which participate in distributed
-# consensus.
-[broker]
-enabled = true
-# Where the Raft logs are stored. The user running InfluxDB will need read/write access.
-dir  = "/var/opt/influxdb/raft"
-truncation-interval = "10m"
-max-topic-size = 52428800
-max-segment-size = 10485760
-
-# Raft configuration. Controls the distributed consensus system.
-[raft]
-apply-interval = "10ms"
-election-timeout = "5s"
-heartbeat-interval = "100ms"
-reconnect-timeout = "10ms"
-
-# Data node configuration. Data nodes are where the time-series data, in the form of
-# shards, is stored.
-[data]
-enabled = true
-dir = "/var/opt/influxdb/db"
-
-# Auto-create a retention policy when a database is created. Defaults to true.
-retention-auto-create = true
-
-# Control whether retention policies are enforced and how long the system waits between
-# enforcing those policies.
-retention-check-enabled = true
-retention-check-period = "10m"
-
-# Configuration for snapshot endpoint.
-[snapshot]
-enabled = true # Enabled by default if not set.
-
-[logging]
-write-tracing = false # If true, enables detailed logging of the write system.
-raft-tracing = false # If true, enables detailed logging of Raft consensus.
-http-access = true # If true, logs each HTTP access to the system.
-
-# InfluxDB can store statistical and diagnostic information about itself. This is useful for
-# monitoring purposes. This feature is disabled by default, but if enabled, these data can be
-# queried like any other data.
 [monitoring]
   enabled = false
-write-interval = "1m"          # Period between writing the data.
+  write-interval = "1m0s"
+
+[continuous_queries]
+  enabled = true
+  recompute-previous-n = 2
+  recompute-no-older-than = "10m0s"
+  compute-runs-per-interval = 10
+  compute-no-more-than = "2m0s"
+
+[hinted-handoff]
+  enabled = true
+  dir = "/var/opt/influxdb/hh"
+  max-size = 1073741824
+  max-age = "168h0m0s"
+  retry-rate-limit = 0
+  retry-interval = "1s"
+

src/InfluxDB/Adapter/AdapterAbstract.php

@@ -28,6 +28,7 @@ abstract class AdapterAbstract implements AdapterInterface
         return [
             "database" => $this->getOptions()->getDatabase(),
             "retentionPolicy" => $this->getOptions()->getRetentionPolicy(),
+            "tags" => $this->getOptions()->getTags(),
         ];
     }
 

src/InfluxDB/Adapter/GuzzleAdapter.php

@@ -25,6 +25,10 @@ class GuzzleAdapter extends AdapterAbstract implements QueryableInterface
     {
         $message = array_replace_recursive($this->getMessageDefaults(), $message);
 
+        if (!count($message["tags"])) {
+            unset($message["tags"]);
+        }
+
         $httpMessage = [
             "auth" => [$this->getOptions()->getUsername(), $this->getOptions()->getPassword()],
             "body" => json_encode($message)

src/InfluxDB/Adapter/UdpAdapter.php

@@ -8,10 +8,67 @@ final class UdpAdapter extends AdapterAbstract
     public function send(array $message)
     {
         $message = array_replace_recursive($this->getMessageDefaults(), $message);
+        $message = $this->serialize($message);
 
-        $message = json_encode($message);
+        $this->write($message);
+    }
+
+    public function write($message)
+    {
         $socket = socket_create(AF_INET, SOCK_DGRAM, SOL_UDP);
         socket_sendto($socket, $message, strlen($message), 0, $this->getOptions()->getHost(), $this->getOptions()->getPort());
         socket_close($socket);
     }
+
+    private function serialize(array $message)
+    {
+        $tags = $this->getOptions()->getTags();
+
+        if (array_key_exists("tags", $message)) {
+            $tags = array_replace_recursive($tags, $message["tags"]);
+        }
+
+        $unixepoch = (int)microtime(true);
+        if (array_key_exists("time", $message)) {
+            $dt = new \DateTime($message["time"]);
+            $unixepoch = $dt->format("U");
+        }
+
+        $lines = [];
+        foreach ($message["points"] as $point) {
+            if (array_key_exists("tags", $point)) {
+                $tags = array_replace_recursive($tags, $point["tags"]);
+            }
+
+            if (!$tags) {
+                $lines[] = trim(
+                    sprintf(
+                        "%s %s %d",
+                        $point["measurement"], $this->toKeyValue($point["fields"], true), $unixepoch
+                    )
+                );
+            } else {
+                $lines[] = trim(
+                    sprintf(
+                        "%s,%s %s %d",
+                        $point["measurement"], $this->toKeyValue($tags), $this->toKeyValue($point["fields"], true), $unixepoch
+                    )
+                );
+            }
+        }
+        return implode("\n", $lines);
+    }
+
+    private function toKeyValue(array $elems, $escape=false)
+    {
+        $list = [];
+        foreach ($elems as $key => $value) {
+            if ($escape && is_string($value)) {
+                $value = "\"{$value}\"";
+            }
+            $list[] = sprintf("%s=%s", $key, $value);
+        }
+
+        return implode(",", $list);
+    }
 }

src/InfluxDB/Client.php

@@ -29,7 +29,7 @@ class Client
         if (!is_array($name)) {
             $data =[];
 
-            $data['points'][0]['name'] = $name;
+            $data['points'][0]['measurement'] = $name;
             $data['points'][0]['fields'] = $values;
         }
 

src/InfluxDB/Options.php

@@ -7,41 +7,22 @@ namespace InfluxDB;
  */
 class Options
 {
-    /**
-     * @var string
-     */
     private $host;
 
-    /**
-     * @var string|int
-     */
     private $port;
 
-    /**
-     * @var string
-     */
     private $username;
 
-    /**
-     * @var string
-     */
     private $password;
 
-    /**
-     * @var string
-     */
     private $protocol;
 
-    /**
-     * @var string
-     */
     private $database;
 
     private $retentionPolicy;
 
-    /**
+    private $tags;
-     * Set default options
+
-     */
     public function __construct()
     {
         $this->setHost("localhost");
@@ -51,6 +32,18 @@ class Options
         $this->setProtocol("http");
 
         $this->setRetentionPolicy("default");
+        $this->setTags([]);
+    }
+
+    public function getTags()
+    {
+        return $this->tags;
+    }
+
+    public function setTags($tags)
+    {
+        $this->tags = $tags;
+        return $this;
     }
 
     public function getRetentionPolicy()

tests/InfluxDB/Adapter/UdpAdapterTest.php

@@ -0,0 +1,64 @@
+<?php
+namespace InfluxDB\Adapter;
+
+use InfluxDB\Options;
+
+class UdpAdapterTest extends \PHPUnit_Framework_TestCase
+{
+    /**
+     * @dataProvider getMessages
+     */
+    public function testRewriteMessages($input, $response)
+    {
+        $object = new UdpAdapter(new Options());
+        $reflection = new \ReflectionClass(get_class($object));
+        $method = $reflection->getMethod("serialize");
+        $method->setAccessible(true);
+
+        $message = $method->invokeArgs($object, [$input]);
+
+        $this->assertEquals($response, $message);
+    }
+
+    public function getMessages()
+    {
+        return [
+            [
+                [
+                    "time" => "2009-11-10T23:00:00Z",
+                    "points" => [
+                        [
+                            "measurement" => "cpu",
+                            "fields" => [
+                                "value" => 1,
+                            ],
+                        ],
+                    ],
+                ],
+                "cpu value=1 1257894000"
+            ],
+            [
+                [
+                    "tags" => [
+                        "region" => "us-west",
+                        "host" => "serverA",
+                        "env" => "prod",
+                        "target" => "servers",
+                        "zone" => "1c",
+                    ],
+                    "time" => "2009-11-10T23:00:00Z",
+                    "points" => [
+                        [
+                            "measurement" => "cpu",
+                            "fields" => [
+                                "cpu" => 18.12,
+                                "free" => 712432,
+                            ],
+                        ],
+                    ],
+                ],
+                "cpu,region=us-west,host=serverA,env=prod,target=servers,zone=1c cpu=18.12,free=712432 1257894000"
+            ]
+        ];
+    }
+}

tests/InfluxDB/ClientTest.php

@@ -108,7 +108,7 @@ class ClientTest extends \PHPUnit_Framework_TestCase
             ],
             "points" => [
                 [
-                    "name" => "vm-serie",
+                    "measurement" => "vm-serie",
                     "fields" => [
                         "cpu" => 18.12,
                         "free" => 712423,
@@ -126,6 +126,37 @@ class ClientTest extends \PHPUnit_Framework_TestCase
         $this->assertEquals(18.12, $body["results"][0]["series"][0]["values"][0][1]);
     }
 
+    /**
+     * @group tcp
+     */
+    public function testOverrideDatabaseNameViaMessage()
+    {
+        $this->options->setDatabase("a-wrong-database");
+
+        $this->object->mark([
+            "database" => "tcp.test",
+            "points" => [
+                [
+                    "measurement" => "vm-serie",
+                    "fields" => [
+                        "cpu" => 18.12,
+                        "free" => 712423,
+                    ],
+                ],
+            ]
+        ]);
+
+        sleep(1);
+
+        $this->options->setDatabase("tcp.test");
+        $body = $this->object->query("select * from \"vm-serie\"");
+
+        $this->assertCount(1, $body["results"][0]["series"][0]["values"]);
+        $this->assertEquals("cpu", $body["results"][0]["series"][0]["columns"][1]);
+        $this->assertEquals(18.12, $body["results"][0]["series"][0]["values"][0][1]);
+        $this->assertEquals(712423, $body["results"][0]["series"][0]["values"][0][2]);
+    }
+
     /**
      * @group udp
      */
@@ -144,60 +175,20 @@ class ClientTest extends \PHPUnit_Framework_TestCase
         $object->setAdapter($adapter);
 
         $object->mark("udp.test", ["mark" => "element"]);
+        sleep(1);
         $object->mark("udp.test", ["mark" => "element1"]);
+        sleep(1);
         $object->mark("udp.test", ["mark" => "element2"]);
+        sleep(1);
         $object->mark("udp.test", ["mark" => "element3"]);
 
         // Wait UDP/IP message arrives
-        sleep(1);
+        sleep(2);
 
         $this->options->setDatabase("udp.test");
         $body = $this->object->query("select * from \"udp.test\"");
 
         $this->assertCount(4, $body["results"][0]["series"][0]["values"]);
-        $this->assertEquals("mark", $body["results"][0]["series"][0]["columns"][1]);
-        $this->assertEquals("element", $body["results"][0]["series"][0]["values"][0][1]);
-    }
-
-    /**
-     * @group udp
-     */
-    public function testOverrideDatabaseNameViaMessage()
-    {
-        $rawOptions = $this->rawOptions;
-        $options = new Options();
-        $options->setHost($rawOptions["udp"]["host"]);
-        $options->setUsername($rawOptions["udp"]["username"]);
-        $options->setPassword($rawOptions["udp"]["password"]);
-        $options->setPort($rawOptions["udp"]["port"]);
-        $options->setDatabase("a-wrong-database");
-
-        $adapter = new UdpAdapter($options);
-        $object = new Client();
-        $object->setAdapter($adapter);
-
-        $object->mark([
-            "database" => "{$rawOptions["udp"]["database"]}",
-            "points" => [
-                [
-                    "name" => "vm-serie",
-                    "fields" => [
-                        "cpu" => 18.12,
-                        "free" => 712423,
-                    ],
-                ],
-            ]
-        ]);
-
-        sleep(1);
-
-        $this->options->setDatabase($rawOptions["udp"]["database"]);
-        $body = $this->object->query("select * from \"vm-serie\"");
-
-        $this->assertCount(1, $body["results"][0]["series"][0]["values"]);
-        $this->assertEquals("cpu", $body["results"][0]["series"][0]["columns"][1]);
-        $this->assertEquals(18.12, $body["results"][0]["series"][0]["values"][0][1]);
-        $this->assertEquals(712423, $body["results"][0]["series"][0]["values"][0][2]);
     }
 
     /**
@@ -220,7 +211,7 @@ class ClientTest extends \PHPUnit_Framework_TestCase
         $object->mark([
             "points" => [
                 [
-                    "name" => "vm-serie",
+                    "measurement" => "vm-serie",
                     "fields" => [
                         "cpu" => 18.12,
                         "free" => 712423,
@@ -229,7 +220,7 @@ class ClientTest extends \PHPUnit_Framework_TestCase
             ]
         ]);
 
-        sleep(1);
+        sleep(2);
 
         $this->options->setDatabase("udp.test");
         $body = $this->object->query("select * from \"vm-serie\"");
@@ -264,7 +255,7 @@ class ClientTest extends \PHPUnit_Framework_TestCase
             ],
             "points" => [
                 [
-                    "name" => "vm-serie",
+                    "measurement" => "vm-serie",
                     "tags" => [
                         "dc"  => "eu-west-1",
                         "one"  => "two",
@@ -275,7 +266,7 @@ class ClientTest extends \PHPUnit_Framework_TestCase
                     ],
                 ],
                 [
-                    "name" => "vm-serie",
+                    "measurement" => "vm-serie",
                     "tags" => [
                         "dc"  => "us-east-1",
                     ],
@@ -287,7 +278,7 @@ class ClientTest extends \PHPUnit_Framework_TestCase
             ]
         ]);
 
-        sleep(1);
+        sleep(2);
 
         $this->options->setDatabase("udp.test");
         $body = $this->object->query("select * from \"vm-serie\" where dc='eu-west-1'");