inotify log file broadcaster

iocBoot/ioccaputlog/Makefile

@@ -0,0 +1,5 @@
+TOP = ../..
+include $(TOP)/configure/CONFIG
+ARCH = linux-x86_64
+TARGETS = envPaths
+include $(TOP)/configure/RULES.ioc

iocBoot/ioccaputlog/st.cmd

@@ -0,0 +1,12 @@
+#!../../bin/linux-x86_64/softIocPy
+
+< envPaths
+
+dbLoadDatabase("../../dbd/softIocPy.dbd",0,0)
+softIocPy_registerRecordDeviceDriver(pdbbase) 
+
+dbLoadRecords("../../db/logwatch.db","N=logrec,FNAME=/tmp/testlog")
+
+iocInit()
+
+dbl > records.dbl

logApp/Db/Makefile

@@ -0,0 +1,22 @@
+TOP=../..
+include $(TOP)/configure/CONFIG
+#----------------------------------------
+#  ADD MACRO DEFINITIONS AFTER THIS LINE
+
+#----------------------------------------------------
+#  Optimization of db files using dbst (DEFAULT: NO)
+#DB_OPT = YES
+
+#----------------------------------------------------
+# Create and install (or just install) into <top>/db
+# databases, templates, substitutions like this
+DB += logwatch.db
+
+#----------------------------------------------------
+# If <anyname>.db template is not named <anyname>*.template add
+# <anyname>_template = <templatename>
+
+include $(TOP)/configure/RULES
+#----------------------------------------
+#  ADD RULES AFTER THIS LINE
+

logApp/Db/logwatch.db

@@ -0,0 +1,9 @@
+
+record(waveform, "$(N)") {
+  field(DTYP, "Python Device")
+  field(INP , "@logwatch $(FNAME)")
+  field(SCAN, "I/O Intr")
+  field(FTVL, "CHAR")
+  field(NELM, "$(NELM=200)")
+  info("logfilter","$(FILTER=)")
+}

logApp/Makefile

@@ -0,0 +1,8 @@
+TOP = ..
+include $(TOP)/configure/CONFIG
+DIRS := $(DIRS) $(filter-out $(DIRS), $(wildcard *src*))
+DIRS := $(DIRS) $(filter-out $(DIRS), $(wildcard *Src*))
+DIRS := $(DIRS) $(filter-out $(DIRS), $(wildcard *db*))
+DIRS := $(DIRS) $(filter-out $(DIRS), $(wildcard *Db*))
+include $(TOP)/configure/RULES_DIRS
+

logApp/src/Makefile

@@ -0,0 +1,24 @@
+TOP=../..
+
+include $(TOP)/configure/CONFIG
+include $(TOP)/configure/CONFIG_PY
+#----------------------------------------
+#  ADD MACRO DEFINITIONS AFTER THIS LINE
+#=============================
+
+#=============================
+# Build the IOC application
+
+LOADABLE_LIBRARY_HOST = _inotifyy
+
+_inotifyy_SRCS += inotify_wrap.c
+
+PY += inotifyy.py
+PY += logwatch.py
+
+#===========================
+
+include $(TOP)/configure/RULES
+include $(TOP)/configure/RULES_PY
+#----------------------------------------
+#  ADD RULES AFTER THIS LINE

logApp/src/inotify_wrap.c

@@ -0,0 +1,281 @@
+
+#include <sys/inotify.h>
+#include <unistd.h>
+#include <fcntl.h>
+
+/* python has its own ideas about which version to support */
+#undef _POSIX_C_SOURCE
+#undef _XOPEN_SOURCE
+
+#include <Python.h>
+#include <structmember.h>
+
+#define EVTMAXSIZE (sizeof(struct inotify_event) + NAME_MAX + 1)
+#define EVTMINSIZE offsetof(struct inotify_event,name)
+
+typedef struct {
+    PyObject_HEAD
+
+    int fd;
+    char buf[EVTMAXSIZE*16];
+} INotify;
+
+static int INotify_Init(INotify *self, PyObject *args, PyObject *kws)
+{
+    int flags;
+    self->fd = inotify_init();
+    if(self->fd==-1) {
+        PyErr_SetFromErrno(PyExc_OSError);
+        return -1;
+    }
+    flags = fcntl(self->fd, F_GETFL, 0);
+    flags |= O_NONBLOCK;
+    if(fcntl(self->fd, F_SETFL, flags)) {
+        close(self->fd);
+        PyErr_SetFromErrno(PyExc_OSError);
+        return -1;
+    }
+    return 0;
+}
+
+static void INotify_dealloc(INotify *self)
+{
+    close(self->fd);
+    self->ob_type->tp_free(self);
+}
+
+static PyObject* INotify_add_watch(INotify* self, PyObject* args)
+{
+    int ret;
+    const char* path;
+    unsigned long mask;
+
+    if(!PyArg_ParseTuple(args, "sk", &path, &mask))
+        return NULL;
+
+    ret = inotify_add_watch(self->fd, path, mask);
+    if(ret==-1) {
+        PyErr_SetFromErrno(PyExc_OSError);
+        return NULL;
+    }
+
+    return PyInt_FromLong(ret);
+}
+
+static PyObject* INotify_rm_watch(INotify* self, PyObject* args)
+{
+    int wd, ret;
+
+    if(!PyArg_ParseTuple(args, "i", &wd))
+        return NULL;
+
+    ret = inotify_rm_watch(self->fd, wd);
+    if(ret==-1) {
+        PyErr_SetFromErrno(PyExc_OSError);
+        return NULL;
+    }
+
+    Py_RETURN_NONE;
+}
+
+/* Reading inotify events (circa. Linux 3.12)
+ * Based on a reading of copy_event_to_user() in inotify_user.c
+ *
+ * The read() call must be given a buffer big enough for at least one
+ * event.  As event size is variable this means the buffer must
+ * be sized for the woust cast (sizeof(inotify_event)+NAME_MAX+1).
+ * We will only be given complete events, but can be sure how many.
+ *
+ * If we don't allocate enough space for one event read() gives EINVAL.
+ * If we don't allocate enough space for two events, read() gives
+ * the size of the first event.
+ * read() should never return zero.
+ */
+
+static PyObject* INotify_read(INotify* self)
+{
+    PyObject *list = NULL;
+    void *buf = self->buf;
+    ssize_t ret;
+
+    list = PyList_New(0);
+    if(!list)
+        return NULL;
+
+retry:
+    ret = read(self->fd, buf, sizeof(self->buf));
+    if(ret<0) {
+        if(errno==EAGAIN)
+            return list; /* return empty list */
+        else if(errno==EINTR) {
+            if(PyErr_CheckSignals()==0)
+                goto retry;
+        }
+        PyErr_SetFromErrno(PyExc_OSError);
+        goto fail;
+    } else if(ret<EVTMINSIZE) {
+        PyErr_Format(PyExc_OSError, "The unthinkable has happened in INotify_read");
+        goto fail;
+    }
+
+    while(ret>=EVTMINSIZE) {
+        PyObject *tuple;
+        struct inotify_event *evt=buf;
+        ssize_t evtsize;
+
+        /* paranoia validation */
+        if(evt->len > ret) {
+            PyErr_Format(PyExc_OSError, "Recieved event length %lu beyond buffer size %lu",
+                         (unsigned long)evt->len, (unsigned long)ret);
+            /* oops, we can't recover from this...  */
+            close(self->fd);
+            self->fd = -1;
+            goto fail;
+        } else if(evt->len>0)
+            evt->name[evt->len-1] = '\0';
+        else
+            evt->name[0] = '\0';
+
+        evtsize = (void*)&evt->name[evt->len] - buf;
+
+        tuple = Py_BuildValue("iIIs",
+                              (int)evt->wd, (unsigned int)evt->mask,
+                              (unsigned int)evt->cookie,
+                              evt->name);
+        if(!tuple)
+            goto fail;
+
+        if(PyList_Append(list, tuple)) {
+            Py_DECREF(tuple);
+            goto fail;
+        }
+
+        buf += evtsize;
+        ret -= evtsize;
+    }
+
+    if(ret!=0)
+        PyErr_Warn(PyExc_UserWarning, "Stray bytes in INotify_read");
+
+    return list;
+fail:
+    Py_XDECREF(list);
+    return NULL;
+}
+
+static struct PyMemberDef INotify_members[] = {
+    {"fd", T_INT, offsetof(INotify, fd), READONLY,
+     "Underlying file descriptor for notifications"},
+    {NULL}
+};
+
+static struct PyMethodDef INotify_methods[] = {
+    {"add", (PyCFunction)INotify_add_watch, METH_VARARGS,
+     "Add a new path to watch"},
+    {"_del", (PyCFunction)INotify_rm_watch, METH_VARARGS,
+     "Stop watching a path"},
+    {"read", (PyCFunction)INotify_read, METH_NOARGS,
+     "Read one event"},
+    {NULL}
+};
+
+static PyTypeObject INotify_type = {
+#if PY_MAJOR_VERSION >= 3
+    PyVarObject_HEAD_INIT(NULL, 0)
+#else
+    PyObject_HEAD_INIT(NULL)
+    0,
+#endif
+    "_inotifyy.INotify",
+    sizeof(INotify),
+};
+
+#if PY_MAJOR_VERSION >= 3
+static struct PyModuleDef inotifymodule = {
+  PyModuleDef_HEAD_INIT,
+    "_inotify",
+    NULL,
+    -1,
+    NULL
+};
+#endif
+
+
+#if PY_MAJOR_VERSION >= 3
+# define MODINIT_RET(VAL) return (VAL)
+#else
+# define MODINIT_RET(VAL) return
+#endif
+
+PyMODINIT_FUNC init_inotifyy(void)
+{
+    PyObject *mod = NULL;
+
+#if PY_MAJOR_VERSION >= 3
+    mod = PyModule_Create(&inotifymodule);
+#else
+    mod = Py_InitModule("_inotifyy", NULL);
+#endif
+    if(!mod)
+        goto fail;
+
+    INotify_type.tp_flags = Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE;
+    INotify_type.tp_members = INotify_members;
+    INotify_type.tp_methods = INotify_methods;
+    INotify_type.tp_init = (initproc)INotify_Init;
+    INotify_type.tp_dealloc = (destructor)INotify_dealloc;
+
+    INotify_type.tp_new = PyType_GenericNew;
+    if(PyType_Ready(&INotify_type)<0) {
+        fprintf(stderr, "INotify object not ready\n");
+        MODINIT_RET(NULL);
+    }
+
+    PyObject *typeobj=(PyObject*)&INotify_type;
+    Py_INCREF(typeobj);
+    if(PyModule_AddObject(mod, "INotify", typeobj)) {
+        Py_DECREF(typeobj);
+        fprintf(stderr, "Failed to add INotify object to module\n");
+        MODINIT_RET(NULL);
+    }
+
+    PyModule_AddIntMacro(mod, IN_ACCESS);
+    PyModule_AddIntMacro(mod, IN_ATTRIB);
+    PyModule_AddIntMacro(mod, IN_CLOSE_WRITE);
+    PyModule_AddIntMacro(mod, IN_CLOSE_NOWRITE);
+    PyModule_AddIntMacro(mod, IN_CREATE);
+    PyModule_AddIntMacro(mod, IN_DELETE);
+    PyModule_AddIntMacro(mod, IN_DELETE_SELF);
+    PyModule_AddIntMacro(mod, IN_MODIFY);
+    PyModule_AddIntMacro(mod, IN_MOVE_SELF);
+    PyModule_AddIntMacro(mod, IN_MOVED_FROM);
+    PyModule_AddIntMacro(mod, IN_MOVED_TO);
+    PyModule_AddIntMacro(mod, IN_OPEN);
+
+    PyModule_AddIntMacro(mod, IN_ALL_EVENTS);
+
+    PyModule_AddIntMacro(mod, IN_EXCL_UNLINK);
+    PyModule_AddIntMacro(mod, IN_ONESHOT);
+    /* added in glibc 2.5 */
+#ifdef IN_DONT_FOLLOW
+    PyModule_AddIntMacro(mod, IN_DONT_FOLLOW);
+#endif
+#ifdef IN_MASK_ADD
+    PyModule_AddIntMacro(mod, IN_MASK_ADD);
+#endif
+#ifdef IN_ONLYDIR
+    PyModule_AddIntMacro(mod, IN_ONLYDIR);
+#endif
+
+    PyModule_AddIntMacro(mod, IN_IGNORED);
+    PyModule_AddIntMacro(mod, IN_ISDIR);
+    PyModule_AddIntMacro(mod, IN_Q_OVERFLOW);
+    PyModule_AddIntMacro(mod, IN_UNMOUNT);
+
+    MODINIT_RET(mod);
+
+fail:
+    fprintf(stderr, "Failed to initialize _inotify module!\n");
+    Py_XDECREF(mod);
+    MODINIT_RET(NULL);
+}

logApp/src/inotifyy.py

@@ -0,0 +1,164 @@
+from __future__ import print_function
+
+# Pull in the IN_* macros
+from _inotifyy import *
+
+import _inotifyy
+import select, os, errno
+
+_flags = {}
+for k in dir(_inotifyy):
+    if k.startswith('IN_') and k!='IN_ALL_EVENTS':
+        _flags[k[3:]] = getattr(_inotifyy, k)
+del k
+
+def decodeMask(mask):
+    ret = []
+    for k,v in _flags.iteritems():
+        if mask&v:
+            ret.append(k)
+    return '%s %s'%(hex(mask),ret)
+
+class IToken(object):
+    def __init__(self, cb, path, inot, wd):
+        self._cb, self._inot, self._wd = cb, inot, wd
+        self.path = path
+    def __del__(self):
+        self.close()
+    def close(self):
+        if self._inot:
+            self._inot._del(self._wd)
+            self._inot = None
+    def __str__(self):
+        return 'IToken(%d,"%s")'%(self._wd, self.path)
+    __repr__=__str__
+
+class INotify(_inotifyy.INotify):
+    def __init__(self):
+        self.__done = False
+        self.__listen, self.__wake = os.pipe()
+        self.__wds = {}
+        super(INotify, self).__init__()
+
+    def close(self):
+        self.__done = True
+        os.write(self.__wake,'*')
+
+    def add(self, callback, path, mask=IN_ALL_EVENTS):
+        wd = super(INotify, self).add(path, mask)
+        try:
+            return self.__wds[wd]
+        except KeyError:
+            tok = IToken(callback, path, self, wd)
+            self.__wds[wd] = tok
+            return tok
+
+    def loop(self):
+        while not self.__done:
+            rds, _wts, _exs = select.select([self.fd, self.__listen], [], [])
+            if self.fd in rds:
+                for wd, mask, cookie, path in self.read():
+                    try:
+                        tok = self.__wds[wd]
+                    except KeyError:
+                        pass # ignore unknown wd
+                    else:
+                        tok._cb(tok, mask, cookie, path)
+            elif self.__listen in rds:
+                os.read(self.__listen,1024)
+        self.__done = False
+
+
+
+def cmdlisten(files):
+    print("Listening for",*files)
+    if len(files)==0:
+        return
+
+    def event(evt, mask, cookie, path):
+        print('Event',evt, decodeMask(mask), cookie, path)
+
+    IN = INotify()
+
+    wds = [IN.add(event, P) for P in files]
+    print(wds)
+
+    return IN
+
+class cmdtail(object):
+    def __init__(self, fname):
+        import os.path
+        self.fname = fname
+        dirname, self.fpart= os.path.split(fname)
+        self.IN = INotify()
+        self.loop = self.IN.loop
+
+        mask=IN_CREATE|IN_DELETE|IN_MOVED_FROM|IN_MODIFY
+
+        self.dirwd = self.IN.add(self.direvt, dirname, mask)
+        self.fd = None
+        self.startfile()
+        self.catfile()
+
+    def startfile(self):
+        self.closefile()
+        try:
+            self.fd = open(self.fname, 'r')
+        except IOError as e:
+            if e.errno==errno.ENOENT:
+                print(self.fname, "Doesn't exist yet")
+                return
+        #self.catfile()
+
+    def closefile(self):
+        if self.fd:
+            print("Closing previous")
+            self.fd.close()
+        self.fd = None
+
+    def catfile(self):
+        if not self.fd:
+           return
+        op = self.fd.tell()
+        self.fd.seek(0, 2)
+        end = self.fd.tell()
+        if end<op:
+            print(self.fname,'got shorter...  Assuming truncation')
+            self.fd.seek(0, 0)
+        else:
+            self.fd.seek(op, 0)
+        while True:
+            D = self.fd.read(1024)
+            if D:
+                print(D)
+            else:
+                break
+
+    def direvt(self, evt, mask, cookie, path):
+        if path!=self.fpart:
+            return
+        print('Dir event',evt, decodeMask(mask), cookie, path)
+        if mask&IN_CREATE:
+            print(self.fname,'appears')
+            self.startfile()
+        if mask&IN_DELETE:
+            print(self.fname,'is deleted')
+            self.closefile()
+        if mask&IN_MOVED_FROM:
+            print(self.fname,'is renamed')
+            self.closefile()
+        if mask&IN_MODIFY:
+            print(self.fname,'is modified')
+            self.catfile()
+
+if __name__=='__main__':
+    import sys
+    if len(sys.argv)<=2:
+        print("Usage: inotifyy listen <path1> [path2] ...")
+        print("    or inotifyy tail <path>")
+        sys.exit(1)
+    elif sys.argv[1]=='listen':
+        IN = cmdlisten(sys.argv[2:])
+    elif sys.argv[1]=='tail':
+        IN = cmdtail(sys.argv[2])
+    IN.loop()

logApp/src/logwatch.py

@@ -0,0 +1,122 @@
+# -*- coding: utf-8 -*-
+from __future__ import print_function
+
+import os.path, errno
+
+import numpy as np
+
+import inotifyy as inot
+
+from devsup.hooks import addHook
+from devsup.util import StoppableThread
+from devsup.db import IOScanListBlock
+
+mask=inot.IN_CREATE|inot.IN_DELETE|inot.IN_MOVED_FROM|inot.IN_MODIFY
+
+class LogWatcher(StoppableThread):
+
+    def __init__(self, rec, args):
+        super(LogWatcher, self).__init__()
+        self.fname = args
+        dir, self.fpart = os.path.split(self.fname)
+        if not os.path.isdir(dir):
+            raise RuntimeError("Directory '%s' does not exist"%dir)
+
+        self.IN = inot.INotify()
+        self.dirwatch = self.IN.add(self.event, dir, mask)
+
+        self.scan = IOScanListBlock()
+        self.allowScan = self.scan.add
+
+        addHook('AfterIocRunning', self.start)
+        addHook('AtIocExit', self.join)
+
+        self.arr = rec.field('VAL').getarray()
+        print('arr',self.arr.dtype)
+        self.fd = None
+        self.buf = None
+        self.msg = ""
+
+        print(rec, 'will watch', self.fname)
+
+    def detach(self, rec):
+        pass
+
+    def process(self, rec, reason):
+        buf = np.frombuffer(self.msg, dtype=self.arr.dtype)
+        buf = buf[:rec.NELM-1]
+        self.arr[:buf.size] = buf
+        self.arr[buf.size] = 0
+        rec.NORD = buf.size+1
+
+    def join(self):
+        print("Stopping logger for",self.fname)
+        self.IN.close()
+        print("Waiting for",self.fname)
+        ret = super(LogWatcher, self).join()
+        print("Joined",self.fname)
+        return ret
+
+    def run(self):
+        print("log watcher staring",self.fname)
+        self.openfile()
+        self.catfile()
+        self.IN.loop()
+
+    def event(self, evt, mask, cookie, path):
+        if path!=self.fpart:
+            return
+
+        if mask&inot.IN_CREATE:
+            self.log('Log file created')
+            self.openfile()
+
+        if mask&(inot.IN_DELETE|inot.IN_MOVED_FROM):
+            self.log("Log file deleted/renamed")
+            self.closefile()
+
+        if mask&(inot.IN_MODIFY):
+            self.catfile()
+
+    def openfile(self):
+        self.closefile()
+        try:
+            self.fd = open(self.fname, 'r')
+        except IOError, e:
+            if e.errno==errno.ENOENT:
+                return
+            raise
+
+    def closefile(self):
+        if self.fd:
+            self.fd.close()
+        self.fd, self.buf = None, None
+
+    def catfile(self):
+        if not self.fd:
+            return
+        op = self.fd.tell()
+        self.fd.seek(0,2) # Seek end
+        end = self.fd.tell()
+        if end < op:
+            self.log("File size decreased, assuming truncation")
+            self.buf = None
+            op = 0
+        self.fd.seek(op,0)
+
+        for L in self.fd.readlines():
+            if L[-1]!='\n':
+                if self.buf:
+                    self.buf += L
+                else:
+                    self.buf = L
+                break
+            elif self.buf:
+                L, self.buf = self.buf+L, None
+            self.log(L[:-1]) # Skip newline
+
+    def log(self, msg):
+        self.msg = msg
+        self.scan.interrupt()
+
+build = LogWatcher