Merge pull request influxdata#159 from influxdata/nc-udf-fixes

udf fixes

Author: Nathaniel Cook

cmd/kapacitor/main.go

@@ -600,7 +600,6 @@ func doDisable(args []string) error {
 		if rp.Error != "" {
 			return errors.New(rp.Error)
 		}
-		return nil
 	}
 	return nil
 }

udf/agent/py/agent.py

@@ -8,6 +8,7 @@
 from threading import Lock, Thread
 from Queue import Queue
 import io
+import traceback
 
 import logging
 logger = logging.getLogger()
@@ -23,19 +24,19 @@
 #
 # To write Points/Batches back to the Agent/Kapacitor use the Agent.write_response method, which is thread safe.
 class Handler(object):
-    def info():
+    def info(self):
         pass
-    def init(init_req):
+    def init(self, init_req):
         pass
-    def snapshot():
+    def snapshot(self):
         pass
-    def restore(restore_req):
+    def restore(self, restore_req):
         pass
-    def begin_batch():
+    def begin_batch(self):
         pass
-    def point():
+    def point(self):
         pass
-    def end_batch(end_req):
+    def end_batch(self, end_req):
         pass
 
 
@@ -122,11 +123,13 @@ def _read_loop(self):
             except EOF:
                 break
             except Exception as e:
+                traceback.print_exc()
                 error = "error processing request of type %s: %s" % (msg, e)
                 logger.error(error)
                 response = udf_pb2.Response()
                 response.error.error = error
                 self.write_response(response)
+                break
 
 # Indicates the end of a file/stream has been reached.
 class EOF(Exception):

udf/agent/py/udf_pb2.py

@@ -167,10 +167,10 @@ message Point {
 // Indicates the end of a batch and contains
 // all meta data associated with the batch.
 message EndBatch {
-    string             Name  = 1;
-    string             Group = 2;
-    int64              TMax  = 3;
-    map<string,string> Tags  = 4;
+    string             name  = 1;
+    string             group = 2;
+    int64              tmax  = 3;
+    map<string,string> tags  = 4;
 }
 
 //-----------------------------------------------------------

udf/udf.pb.go

@@ -460,20 +460,24 @@ func (p *UDFProcess) watchKeepalive() {
 	defer func() {
 		if err != nil {
 			p.setError(err)
+			aborted := make(chan struct{})
 			go func() {
 				timeout := p.keepaliveTimeout * 2
 				if timeout <= 0 {
 					timeout = time.Second
 				}
 				time.Sleep(timeout)
-				p.mu.Lock()
-				defer p.mu.Unlock()
-				if !p.stopped {
+				select {
+				case <-aborted:
+					// We cleanly aborted process is stopped
+				default:
+					// We failed to abort just kill it.
 					p.logger.Println("E! process not responding! killing")
 					p.kill()
 				}
 			}()
 			p.abort()
+			close(aborted)
 		}
 	}()
 	defer p.requestsGroup.Done()
@@ -640,7 +644,10 @@ func (p *UDFProcess) writeBatch(b models.Batch) error {
 
 	req.Message = &udf.Request_End{
 		&udf.EndBatch{
-			Name: b.Name,
+			Name:  b.Name,
+			Group: string(b.Group),
+			Tmax:  b.TMax.UnixNano(),
+			Tags:  b.Tags,
 		},
 	}
 	return p.writeRequest(req)
@@ -746,7 +753,7 @@ func (p *UDFProcess) handleResponse(response *udf.Response) error {
 		}
 	case *udf.Response_End:
 		p.batch.Name = msg.End.Name
-		p.batch.TMax = time.Unix(0, msg.End.TMax)
+		p.batch.TMax = time.Unix(0, msg.End.Tmax)
 		p.batch.Group = models.GroupID(msg.End.Group)
 		p.batch.Tags = msg.End.Tags
 		select {

udf/udf.proto

@@ -5,7 +5,7 @@
 # of structs into property methods and chaining methods.
 
 dest=$1 # output path for the .md files
-docspath=${2-/kapacitor/v0.2/tick}
+docspath=${2-/kapacitor/v0.10/tick}
 
 if [ -z "$dest" ]
 then