Blog Archives

Sending sockets from PostgreSQL triggers with Python

Picture this: We want to notify to one external service each time that one record is inserted in the database. We can find the place where the insert statement is done and create a TCP client there, but: What happens if the application that inserts the data within the database is a legacy application?, or maybe it is too hard to do?. If your database is PostgreSQL it’s pretty straightforward. With the “default” procedural language of PostgreSQL (pgplsql) we cannot do it, but PostgreSQL allows us to use more procedural languages than plpgsql, for example Python. With plpython we can use sockets in the same way than we use it within Python scripts. It’s very simple. Let me show you how to do it.

First we need to create one plpython with our TCP client

CREATE OR REPLACE FUNCTION dummy.sendsocket(msg character varying, host character varying, port integer)
  RETURNS integer AS
$BODY$
  import _socket
  try:
    s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM)
    s.connect((host, port))
    s.sendall(msg)
    s.close()
    return 1
  except:
    return 0
$BODY$
  LANGUAGE plpython VOLATILE
  COST 100;
ALTER FUNCTION dummy.sendsocket(character varying, character varying, integer)
  OWNER TO username;

Now we create the trigger that use our socket client.

CREATE OR REPLACE FUNCTION dummy.myTriggerToSendSockets()
RETURNS trigger AS
$BODY$
   import json
   stmt = plpy.prepare("select dummy.sendSocket($1, $2, $3)", ["text", "text", "int"])
   rv = plpy.execute(stmt, [json.dumps(TD), "host", 26200])
$BODY$
LANGUAGE plpython VOLATILE
COST 100;

As you can see in my example we are sending all the record as a JSON string in the socket body.

And finally we attach the trigger to one table (or maybe we need to do it to more than one table)

CREATE TRIGGER myTrigger
  AFTER INSERT OR UPDATE OR DELETE
  ON dummy.myTable
  FOR EACH ROW
  EXECUTE PROCEDURE dummy.myTriggerToSendSockets();

And that’s all. Now we can use one simple TCP socket server to handle those requests. Let me show you different examples of TCP servers with different languages. As we can see all are different implementations of Reactor pattern. We can use, for example:

node.js:

var net = require('net');

var host = 'localhost';
var port = 26200;

var server = net.createServer(function (socket) {
    socket.on('data', function(buffer) {
        // do whatever that we want with buffer
    });
});

server.listen(port, host);

python (with Twisted):

from twisted.internet import reactor, protocol

HOST = 'localhost'
PORT = 26200

class MyServer(protocol.Protocol):
    def dataReceived(self, data):
        # do whatever that we want with data
        pass

class MyServerFactory(protocol.Factory):
    def buildProtocol(self, addr):
        return MyServer()

reactor.listenTCP(PORT, MyServerFactory(), interface=HOST)
reactor.run()

(I know that we can create the Python’s TCP server without Twisted, but if don’t use it maybe someone will angry with me. Probably he is angry right now because I put the node.js example first :))

php (with react):

<?php
include __DIR__ . '/vendor/autoload.php';

$host = 'localhost';
$port = 26200;

$loop   = React\EventLoop\Factory::create();
$socket = new React\Socket\Server($loop);

$socket->on('connection', function ($conn) {
    $conn->on('data', function ($data) {
        // do whatever we want with data
        }
    );
});

$socket->listen($port, $host);
$loop->run();

You also can use xinet.d to handle the TCP inbound connections.

Building a simple TCP proxy server with node.js

Today we are going to build a simple TCP proxy server. The scenario is the following one. We have got one host (the client) that establishes a TCP connection to another one (the remote).

client —> remote

We want to set up a proxy server in the middle, so the client will establish the connection with the proxy and the proxy will forward it to the remote, keeping in mind the remote response also.
With node.js is really simple to perform those kind of network operations.

client —> proxy -> remote

var net = require('net');

var LOCAL_PORT  = 6512;
var REMOTE_PORT = 6512;
var REMOTE_ADDR = "192.168.1.25";

var server = net.createServer(function (socket) {
    socket.on('data', function (msg) {
        console.log('  ** START **');
        console.log('<< From client to proxy ', msg.toString());
        var serviceSocket = new net.Socket();
        serviceSocket.connect(parseInt(REMOTE_PORT), REMOTE_ADDR, function () {
            console.log('>> From proxy to remote', msg.toString());
            serviceSocket.write(msg);
        });
        serviceSocket.on("data", function (data) {
            console.log('<< From remote to proxy', data.toString());
            socket.write(data);
            console.log('>> From proxy to client', data.toString());
        });
    });
});

server.listen(LOCAL_PORT);
console.log("TCP server accepting connection on port: " + LOCAL_PORT);

Simple, isn’t it?
Source code in github

Talk about node.js and WebSockets

Last friday I spoke about node.js and Websockets with the people of The Mêlée. The talk was an introduction to node.js and focused in the new HTML5 feature: the WebSockets.

When I spoke about Websockets I also introduced the great library socket.io. The jQuery of WebSockets.

Web console with node.js

Continuing with my experiments of node.js, this time I want to create a Web console. The idea is simple. I want to send a few command to the server and I display the output inside the browser. I can do it entirely with PHP but I want to send the output to the browser as fast as they appear without waiting for the end of the command. OK we can do it flushing the output in the server but this solution normally crashes if we keep the application open for a long time. WebSockets again to the rescue. If we need a cross-browser implementation we need the socket.io library. Let’s start:

The node server is a simple websocket server. In this example we will launch each command with spawn function (require(‘child_process’).spawn) and send the output within the websoket. Simple and pretty straightforward.

var sys   = require('sys'),
http  = require('http'),
url   = require('url'),
spawn = require('child_process').spawn,
ws    = require('./ws.js');

var availableComands = ['ls', 'ps', 'uptime', 'tail', 'cat'];
ws.createServer(function(websocket) {
    websocket.on('connect', function(resource) {
        var parsedUrl = url.parse(resource, true);
        var rawCmd = parsedUrl.query.cmd;
        var cmd = rawCmd.split(" ");
        if (cmd[0] == 'help') {
            websocket.write("Available comands: \n");
            for (i=0;i<availableComands.length;i++) {
                websocket.write(availableComands[i]);
                if (i< availableComands.length - 1) {
                    websocket.write(", ");
                }
            }
            websocket.write("\n");

            websocket.end();
        } else if (availableComands.indexOf(cmd[0]) >= 0) {
            if (cmd.length > 1) {
                options = cmd.slice(1);
            } else {
                options = [];
            }
            
            try {
                var process = spawn(cmd[0], options);
            } catch(err) {
                console.log(err);
                websocket.write("ERROR");
            }

            websocket.on('end', function() {
                process.kill();
            });

            process.stdout.on('data', function(data) {
                websocket.write(data);
            });

            process.stdout.on('end', function() {
                websocket.end();
            });
        } else {
             websocket.write("Comand not available. Type help for available comands\n");
             websocket.end();
        }
    });
  
}).listen(8880, '127.0.0.1');

The web client is similar than the example of my previous post

var timeout = 5000;
var wsServer = '127.0.0.1:8880';

var ws;


function cleanString(string) {
    return string.replace(/&/g,"&amp;").replace(/</g,"&lt;").replace(/>/g,"&gt;");
}


function pad(n) {
    return ("0" + n).slice(-2);
}

var cmdHistory = [];
function send(msg) {
    if (msg == 'clear') {
        $('#log').html('');
        return;
    }
    try {
        ws = new WebSocket('ws://' + wsServer + '?cmd=' + msg);
        $('#toolbar').css('background', '#933');
        $('#socketStatus').html("working ... [<a href='#' onClick='quit()'>X</a>]");
        cmdHistory.push(msg);
        $('#log').append("<div class='cmd'>" + msg + "</div>");
        console.log("startWs:");
    } catch (err) {
        console.log(err);
        setTimeout(startWs, timeout);
    }

    ws.onmessage = function(event) {
        $('#log').append(util.toStaticHTML(event.data));
        window.scrollBy(0, 100000000000000000);
    };

    ws.onclose = function(){
        //console.log("ws.onclose");
        $('#toolbar').css('background', '#65A33F');
        $('#socketStatus').html('Type your comand:');
    }
}

function quit() {
    ws.close();
    window.scrollBy(0, 100000000000000000);
}
util = {
  urlRE: /https?:\/\/([-\w\.]+)+(:\d+)?(\/([^\s]*(\?\S+)?)?)?/g, 

  //  html sanitizer 
  toStaticHTML: function(inputHtml) {
    inputHtml = inputHtml.toString();
    return inputHtml.replace(/&/g, "&amp;")
                    .replace(/</g, "&lt;")
                    .replace("/n", "<br/>")
                    .replace(/>/g, "&gt;");
  }, 

  //pads n with zeros on the left,
  //digits is minimum length of output
  //zeroPad(3, 5); returns "005"
  //zeroPad(2, 500); returns "500"
  zeroPad: function (digits, n) {
    n = n.toString();
    while (n.length < digits) 
      n = '0' + n;
    return n;
  },

  //it is almost 8 o'clock PM here
  //timeString(new Date); returns "19:49"
  timeString: function (date) {
    var minutes = date.getMinutes().toString();
    var hours = date.getHours().toString();
    return this.zeroPad(2, hours) + ":" + this.zeroPad(2, minutes);
  },

  //does the argument only contain whitespace?
  isBlank: function(text) {
    var blank = /^\s*$/;
    return (text.match(blank) !== null);
  }
};
$(document).ready(function() {
  //submit new messages when the user hits enter if the message isnt blank
  $("#entry").keypress(function (e) {
    console.log(e);
    if (e.keyCode != 13 /* Return */) return;
    var msg = $("#entry").attr("value").replace("\n", "");
    if (!util.isBlank(msg)) send(msg);
    $("#entry").attr("value", ""); // clear the entry field.
  });
});

And that’s all. In fact we don’t need any line of PHP to perform this web console. Last year I tried to do something similar with PHP but it was a big mess. With node those kind of jobs are trivial. I don’t know if node.js is the future or is just another hype, but it’s easy. And cool. Really cool.

You can see the full code at Github here. Anyway you must take care if you run this application on your host. You are letting user to execute raw unix commands. A bit of security layer would be necessary.

Real time monitoring PHP applications with websockets and node.js

The inspection of the error logs is a common way to detect errors and bugs. We also can show errors on-screen within our developement server, or we even can use great tools like firePHP to show our PHP errors and warnings inside our firebug console. That’s cool, but we only can see our session errors/warnings. If we want to see another’s errors we need to inspect the error log. tail -f is our friend, but we need to surf against all the warnings of all sessions to see our desired ones. Because of that I want to build a tool to monitor my PHP applications in real-time. Let’s start:

What’s the idea? The idea is catch all PHP’s errors and warnings at run time and send them to a node.js HTTP server. This server will work similar than a chat server but our clients will only be able to read the server’s logs. Basically the applications have three parts: the node.js server, the web client (html5) and the server part (PHP). Let me explain a bit each part:

The node Server

Basically it has two parts: a http server to handle the PHP errors/warnings and a websocket server to manage the realtime communications with the browser. When I say that I’m using websockets that’s means the web client will only work with a browser with websocket support like chrome. Anyway it’s pretty straightforward swap from a websocket sever to a socket.io server to use it with every browser. But websockets seems to be the future, so I will use websockets in this example.

The http server:

http.createServer(function (req, res) {
    var remoteAdrress = req.socket.remoteAddress;
    if (allowedIP.indexOf(remoteAdrress) >= 0) {
        res.writeHead(200, {
            'Content-Type': 'text/plain'
        });
        res.end('Ok\n');
        try {
            var parsedUrl = url.parse(req.url, true);
            var type = parsedUrl.query.type;
            var logString = parsedUrl.query.logString;
            var ip = eval(parsedUrl.query.logString)[0];
            if (inspectingUrl == "" ||  inspectingUrl == ip) {
                clients.forEach(function(client) {
                    client.write(logString);
                });
            }
        } catch(err) {
            console.log("500 to " + remoteAdrress);
            res.writeHead(500, {
                'Content-Type': 'text/plain'
            });
            res.end('System Error\n');
        }
    } else {
        console.log("401 to " + remoteAdrress);
        res.writeHead(401, {
            'Content-Type': 'text/plain'
        });
        res.end('Not Authorized\n');
    }
}).listen(httpConf.port, httpConf.host);

and the web socket server:

var inspectingUrl = undefined;

ws.createServer(function(websocket) {
    websocket.on('connect', function(resource) {
        var parsedUrl = url.parse(resource, true);
        inspectingUrl = parsedUrl.query.ip;
        clients.push(websocket);
    });

    websocket.on('close', function() {
        var pos = clients.indexOf(websocket);
        if (pos >= 0) {
            clients.splice(pos, 1);
        }
    });

}).listen(wsConf.port, wsConf.host);

If you want to know more about node.js and see more examples, have a look to the great site: http://nodetuts.com/. In this site Pedro Teixeira will show examples and node.js tutorials. In fact my node.js http + websoket server is a mix of two tutorials from this site.

The web client.

The web client is a simple websockets application. We will handle the websockets connection, reconnect if it dies and a bit more. I’s based on node.js chat demo

<?php $ip = filter_input(INPUT_GET, 'ip', FILTER_SANITIZE_STRING); ?>

        Real time <?= $ip ?> monitor
<script type="text/javascript" src="http://ajax.googleapis.com/ajax/libs/jquery/1.4.3/jquery.min.js"></script><script type="text/javascript">// <![CDATA[
            selectedIp = '<?= $ip ?>';

// ]]></script>
<script type="text/javascript" src="js.js"></script>
</pre>
<div id="toolbar">
<ul id="status">
	<li>Socket status: <span id="socketStatus">Conecting ...</span></li>
	<li>IP: <!--?= $ip == '' ? 'all' : $ip . " <a href='?ip='-->[all]" ?></li>
	<li>count: <span id="count">0</span></li>
</ul>
</div>
<pre>


And the javascript magic

var timeout = 5000;
var wsServer = '192.168.2.2:8880';
var unread = 0;
var focus = false;

var count = 0;
function updateCount() {
    count++;
    $("#count").text(count);
}

function cleanString(string) {
    return string.replace(/&/g,"&amp;").replace(/</g,"&lt;").replace(/>/g,"&gt;");
}

function updateUptime () {
    var now = new Date();
    $("#uptime").text(now.toRelativeTime());
}

function updateTitle(){
    if (unread) {
        document.title = "(" + unread.toString() + ") Real time " + selectedIp + " monitor";
    } else {
        document.title = "Real time " + selectedIp + " monitor";
    }
}

function pad(n) {
    return ("0" + n).slice(-2);
}

function startWs(ip) {
    try {
        ws = new WebSocket("ws://" + wsServer + "?ip=" + ip);
        $('#toolbar').css('background', '#65A33F');
        $('#socketStatus').html('Connected to ' + wsServer);
        //console.log("startWs:" + ip);
        //listen for browser events so we know to update the document title
        $(window).bind("blur", function() {
            focus = false;
            updateTitle();
        });

        $(window).bind("focus", function() {
            focus = true;
            unread = 0;
            updateTitle();
        });
    } catch (err) {
        //console.log(err);
        setTimeout(startWs, timeout);
    }

    ws.onmessage = function(event) {
        unread++;
        updateTitle();
        var now = new Date();
        var hh = pad(now.getHours());
        var mm = pad(now.getMinutes());
        var ss = pad(now.getSeconds());

        var timeMark = '[' + hh + ':' + mm + ':' + ss + '] ';
        logString = eval(event.data);
        var host = logString[0];
        var line = "<table class='message'><tr><td width='1%' class='date'>" + timeMark + "</td><td width='1%' valign='top' class='host'><a href=?ip=" + host + ">" + host + "</a></td>";
        line += "<td class='msg-text' width='98%'>" + logString[1]; + "</td></tr>";
        if (logString[2]) {
            line += "<tr><td>&nbsp;</td><td colspan='3' class='msg-text'>" + logString[2] + "</td></tr>";
        }

        $('#log').append(line);
        updateCount();
        window.scrollBy(0, 100000000000000000);
    };

    ws.onclose = function(){
        //console.log("ws.onclose");
        $('#toolbar').css('background', '#933');
        $('#socketStatus').html('Disconected');
        setTimeout(function() {startWs(selectedIp)}, timeout);
    }
}

$(document).ready(function() {
    startWs(selectedIp);
});

The server part:

The server part will handle silently all PHP warnings and errors and it will send them to the node server. The idea is to place a minimal PHP line of code at the beginning of the application that we want to monitor. Imagine the following piece of PHP code

$a = $var[1];
$a = 1/0;
class Dummy
{
    static function err()
    {
        throw new Exception("error");
    }
}
Dummy1::err();

it will throw:
A notice: Undefined variable: var
A warning: Division by zero
An Uncaught exception ‘Exception’ with message ‘error’

So we will add our small library to catch those errors and send them to the node server

include('client/NodeLog.php');
NodeLog::init('192.168.2.2');

$a = $var[1];
$a = 1/0;
class Dummy
{
    static function err()
    {
        throw new Exception("error");
    }
}
Dummy1::err();

The script will work in the same way than the fist version but if we start our node.js server in a console:

$ node server.js
HTTP server started at 192.168.2.2::5672
Web Socket server started at 192.168.2.2::8880

We will see those errors/warnings in real-time when we start our browser

Here we can see a small screencast with the working application:

This is the server side library:

class NodeLog
{
    const NODE_DEF_HOST = '127.0.0.1';
    const NODE_DEF_PORT = 5672;

    private $_host;
    private $_port;

    /**
     * @param String $host
     * @param Integer $port
     * @return NodeLog
     */
    static function connect($host = null, $port = null)
    {
        return new self(is_null($host) ? self::$_defHost : $host, is_null($port) ? self::$_defPort : $port);
    }

    function __construct($host, $port)
    {
        $this->_host = $host;
        $this->_port = $port;
    }

    /**
     * @param String $log
     * @return Array array($status, $response)
     */
    public function log($log)
    {
        list($status, $response) = $this->send(json_encode($log));
        return array($status, $response);
    }

    private function send($log)
    {
        $url = "http://{$this->_host}:{$this->_port}?logString=" . urlencode($log);
        $ch = curl_init();
        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_NOBODY, true);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);

        $response = curl_exec($ch);
        $status   = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        curl_close($ch);

        return array($status, $response);
    }

    static function getip() {
        $realip = '0.0.0.0';
        if ($_SERVER) {
            if ( isset($_SERVER['HTTP_X_FORWARDED_FOR']) && $_SERVER['HTTP_X_FORWARDED_FOR'] ) {
                $realip = $_SERVER["HTTP_X_FORWARDED_FOR"];
            } elseif ( isset($_SERVER['HTTP_CLIENT_IP']) && $_SERVER["HTTP_CLIENT_IP"] ) {
                $realip = $_SERVER["HTTP_CLIENT_IP"];
            } else {
                $realip = $_SERVER["REMOTE_ADDR"];
            }
        } else {
            if ( getenv('HTTP_X_FORWARDED_FOR') ) {
                $realip = getenv('HTTP_X_FORWARDED_FOR');
            } elseif ( getenv('HTTP_CLIENT_IP') ) {
                $realip = getenv('HTTP_CLIENT_IP');
            } else {
                $realip = getenv('REMOTE_ADDR');
            }
        }
        return $realip;
    }

    public static function getErrorName($err)
    {
        $errors = array(
            E_ERROR             => 'ERROR',
            E_RECOVERABLE_ERROR => 'RECOVERABLE_ERROR',
            E_WARNING           => 'WARNING',
            E_PARSE             => 'PARSE',
            E_NOTICE            => 'NOTICE',
            E_STRICT            => 'STRICT',
            E_DEPRECATED        => 'DEPRECATED',
            E_CORE_ERROR        => 'CORE_ERROR',
            E_CORE_WARNING      => 'CORE_WARNING',
            E_COMPILE_ERROR     => 'COMPILE_ERROR',
            E_COMPILE_WARNING   => 'COMPILE_WARNING',
            E_USER_ERROR        => 'USER_ERROR',
            E_USER_WARNING      => 'USER_WARNING',
            E_USER_NOTICE       => 'USER_NOTICE',
            E_USER_DEPRECATED   => 'USER_DEPRECATED',
        );
        return $errors[$err];
    }

    private static function set_error_handler($nodeHost, $nodePort)
    {
        set_error_handler(function ($errno, $errstr, $errfile, $errline) use($nodeHost, $nodePort) {
            $err = NodeLog::getErrorName($errno);
            /*
            if (!(error_reporting() & $errno)) {
                // This error code is not included in error_reporting
                return;
            }
            */
            $log = array(
                NodeLog::getip(),
                "<strong class="{$err}">{$err}</strong> {$errfile}:{$errline}",
                nl2br($errstr)
            );
            NodeLog::connect($nodeHost, $nodePort)->log($log);
            return false;
        });
    }

    private static function register_exceptionHandler($nodeHost, $nodePort)
    {
        set_exception_handler(function($exception) use($nodeHost, $nodePort) {
            $exceptionName = get_class($exception);
            $message = $exception->getMessage();
            $file = $exception->getFile();
            $line = $exception->getLine();
            $trace = $exception->getTraceAsString();

            $msg = count($trace) > 0 ? "Stack trace:\n{$trace}" : null;
            $log = array(
                NodeLog::getip(),
                nl2br("<strong class="ERROR">Uncaught exception '{$exceptionName}'</strong> with message '{$message}' in {$file}:{$line}"),
                nl2br($msg)
            );
            NodeLog::connect($nodeHost, $nodePort)->log($log);
            return false;
        });
    }

    private static function register_shutdown_function($nodeHost, $nodePort)
    {
        register_shutdown_function(function() use($nodeHost, $nodePort) {
            $error = error_get_last();

            if ($error['type'] == E_ERROR) {
                $err = NodeLog::getErrorName($error['type']);
                $log = array(
                    NodeLog::getip(),
                    "<strong class="{$err}">{$err}</strong> {$error['file']}:{$error['line']}",
                    nl2br($error['message'])
                );
                NodeLog::connect($nodeHost, $nodePort)->log($log);
            }
            echo NodeLog::connect($nodeHost, $nodePort)->end();
        });
    }

    private static $_defHost = self::NODE_DEF_HOST;
    private static $_defPort = self::NODE_DEF_PORT;

    /**
     * @param String $host
     * @param Integer $port
     * @return NodeLog
     */
    public static function init($host = self::NODE_DEF_HOST, $port = self::NODE_DEF_PORT)
    {
        self::$_defHost = $host;
        self::$_defPort = $port;

        self::register_exceptionHandler($host, $port);
        self::set_error_handler($host, $port);
        self::register_shutdown_function($host, $port);

        $node = self::connect($host, $port);
        $node->start();
        return $node;
    }

    private static $time;
    private static $mem;

    public function start()
    {
        self::$time = microtime(TRUE);
        self::$mem = memory_get_usage();
        $log = array(NodeLog::getip(), "<strong class="OK">Start</strong> >>>> {$_SERVER['REQUEST_URI']}");
        $this->log($log);
    }

    public function end()
    {
        $mem = (memory_get_usage() - self::$mem) / (1024 * 1024);
        $time = microtime(TRUE) - self::$time;
        $log = array(NodeLog::getip(), "<strong class="OK">End</strong> <<<< mem: {$mem} time {$time}");         $this->log($log);
    }
}

And of course the full code on gitHub: RealTimeMonitor

Follow

Get every new post delivered to your Inbox.

Join 976 other followers