Merge pull request #29 from BenB196/staging

Merge staging into master and bump to v0.1.0

README.md

@@ -104,7 +104,7 @@ Currently, only JSON formatted configuration files are accepted, in the future Y
         }
       ]
     },
-    "outputType": "elastic",                                                                                                #Output type, supports either file or elastic
+    "outputType": "elastic",                                                                                                #Output type, supports either file, elastic, logstash
     "outputLocation": "/path/to/output",                                                                                    #This is needed even if not using file output type, as there are stateful files which need to be written and stored.
     "ip-api": {                                                                                                             #IP-API Integration
       "enabled": true,                                                                                                      #Enable IP-API support? Default = false
@@ -130,6 +130,9 @@ Currently, only JSON formatted configuration files are accepted, in the future Y
       },
       "aliases": ["test1","test2"]                                                                                          #Any aliases you want the index to be created with.
     }
+    "logstash": {                                                                                                           #Logstash output
+      "logstashURL": "192.168.1.105:8080"                                                                                   #Address of logstash
+    }
   },
     {
       "name": "example_query_2",                                                                                            #Second example FFS query, with extremely simple setup.
@@ -205,6 +208,20 @@ If you are using the elastic output type there are a few important things to und
       1. bestCompression
       1. refreshInterval
       1. aliases
+
+### Logstash Integration
+
+Even though this is for Logstash integration, it uses a standard TCP socket output, so in theory this can be integrated with anything that accepts TCP data.
+
+Logstash pipeline input:
+
+```
+input {
+  tcp {
+    port => 8080 #Set your port
+  }
+}
+```
 
 ### IP-API Integration
 

VERSION

@@ -1 +1 @@
-0.0.9
+0.1.0

config/configReader.go

@@ -34,6 +34,7 @@ type FFSQuery struct {
 	OutputLocation  string			`json:"outputLocation,omitempty"`
 	IPAPI			IPAPI			`json:"ip-api,omitempty"`
 	Elasticsearch	Elasticsearch	`json:"elasticsearch,omitempty"`
+	Logstash		Logstash		`json:"logstash,omitempty"`
 }
 
 type IPAPI struct {
@@ -59,6 +60,10 @@ type Elasticsearch struct {
 	Aliases					[]string	`json:"aliases,omitempty"`
 }
 
+type Logstash struct {
+	LogstashURL string 	`json:"logstashURL"`
+}
+
 type BasicAuth struct {
 	User			string			`json:"user,omitempty"`
 	Password		string			`json:"password,omitempty"`
@@ -377,7 +382,10 @@ func validateConfigJson(fileBytes []byte) (Config, error) {
 							}
 						}
 					}
-
+				case "logstash":
+					if query.Logstash.LogstashURL == "" {
+						return config, errors.New("error: logstash url cannot be blank")
+					}
 				default:
 					return config, errors.New("unknown output type provide in ffs query: " + query.Name + ", output type provided: " + query.OutputType)
 				}

elasticsearch/logstash.go

@@ -0,0 +1,26 @@
+package elasticsearch
+
+import (
+	"net"
+)
+
+func CreateLogstashClient(logstashURL string) (net.Conn,error) {
+	tcpAddr, err := net.ResolveTCPAddr("tcp",logstashURL)
+
+	if err != nil {
+		return nil, err
+	}
+
+	connection, err := net.DialTCP("tcp",nil,tcpAddr)
+
+	if err != nil {
+		return nil, err
+	}
+	err = connection.SetWriteBuffer(100000)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return connection, nil
+}

ffsEvent/ffsEvent.go

@@ -1,6 +1,7 @@
 package ffsEvent
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"errors"
@@ -488,6 +489,65 @@ func queryFetcher(query config.FFSQuery, inProgressQueries *[]eventOutput.InProg
 				log.Println("error closing elastic bulk request")
 				panic(err)
 			}
+		case "logstash":
+			var logstashWg sync.WaitGroup
+
+			conn, err := elasticsearch.CreateLogstashClient(query.Logstash.LogstashURL)
+
+			if err != nil {
+				//TODO handle error
+				log.Println("error creating logstash connection")
+				panic(err)
+			}
+
+			writer := bufio.NewWriter(conn)
+
+			logstashWg.Add(len(ffsEvents))
+			go func() {
+				for _, ffsEvent := range ffsEvents {
+					event, err := json.Marshal(ffsEvent)
+
+					if err != nil {
+						//TODO handle error
+						log.Println("error marshaling ffs event")
+						log.Println(ffsEvent)
+						panic(err)
+					}
+
+					_, err = writer.Write(event)
+
+					if err != nil {
+						//TODO handle error
+						log.Println("error writing ffs event")
+						log.Println(string(event))
+						panic(err)
+					}
+					_, err = writer.Write([]byte("\n"))
+					if err != nil {
+						//TODO handle error
+						log.Println("error writing ffs event")
+						log.Println(string(event))
+						panic(err)
+					}
+					logstashWg.Done()
+				}
+			}()
+
+			logstashWg.Wait()
+
+			err = writer.Flush()
+
+			if err != nil {
+				log.Println("error flushing logstash buffer")
+				panic(err)
+			}
+
+			err = conn.Close()
+
+			if err != nil {
+				log.Println("error closing logstash connection")
+				panic(err)
+			}
 		}
 	}
 

go.mod

@@ -3,9 +3,9 @@ module github.com/BenB196/crashplan-ffs-puller
 go 1.12
 
 require (
-	github.com/BenB196/crashplan-ffs-go-pkg v0.0.5
+	github.com/BenB196/crashplan-ffs-go-pkg v0.0.6
 	github.com/BenB196/ip-api-go-pkg v0.0.4
-	github.com/google/go-cmp v0.3.0
+	github.com/google/go-cmp v0.3.1
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/olivere/elastic/v7 v7.0.6

go.sum

@@ -10,6 +10,8 @@ github.com/BenB196/crashplan-ffs-go-pkg v0.0.4 h1:LBIAOVn9HXiJSpA1xoVEIF9CI0HhGd
 github.com/BenB196/crashplan-ffs-go-pkg v0.0.4/go.mod h1:oBwZ5/KfaNokjvI2Hn2e6c5801ir11FOCq4lG5PcniE=
 github.com/BenB196/crashplan-ffs-go-pkg v0.0.5 h1:ioDSdeFzIc7351h95qwTbL6tXkinDSiYImVMgLRtkoA=
 github.com/BenB196/crashplan-ffs-go-pkg v0.0.5/go.mod h1:oBwZ5/KfaNokjvI2Hn2e6c5801ir11FOCq4lG5PcniE=
+github.com/BenB196/crashplan-ffs-go-pkg v0.0.6 h1:kZzkUpz2epVIgJJZSEsBYdEW2OhB2QfPLvtW7peP4WA=
+github.com/BenB196/crashplan-ffs-go-pkg v0.0.6/go.mod h1:oBwZ5/KfaNokjvI2Hn2e6c5801ir11FOCq4lG5PcniE=
 github.com/BenB196/ip-api-go-pkg v0.0.3 h1:FWmM7FkhT1N55jd4jPW7W9LVOQrG89DLlrgwMb5fosw=
 github.com/BenB196/ip-api-go-pkg v0.0.3/go.mod h1:ccPdkBNnzf/uvuk7qXgEO06TCS/qILNJQP/KETQG4jU=
 github.com/BenB196/ip-api-go-pkg v0.0.4 h1:BbWELxooG6l2gaXQ4i4gm6NsyikdbhalF+TKg08gDPQ=
@@ -50,6 +52,8 @@ github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8l
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.3.1 h1:Xye71clBPdm5HgqGwUkwhbynsUJZhDbS20FvLhQ2izg=
+github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/gorilla/context v1.1.1/go.mod h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg=
 github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=

vendor/modules.txt

@@ -1,12 +1,12 @@
-# github.com/BenB196/crashplan-ffs-go-pkg v0.0.5
+# github.com/BenB196/crashplan-ffs-go-pkg v0.0.6
 github.com/BenB196/crashplan-ffs-go-pkg
 # github.com/BenB196/ip-api-go-pkg v0.0.4
 github.com/BenB196/ip-api-go-pkg
 # github.com/beorn7/perks v1.0.1
 github.com/beorn7/perks/quantile
 # github.com/golang/protobuf v1.3.2
 github.com/golang/protobuf/proto
-# github.com/google/go-cmp v0.3.0
+# github.com/google/go-cmp v0.3.1
 github.com/google/go-cmp/cmp
 github.com/google/go-cmp/cmp/internal/diff
 github.com/google/go-cmp/cmp/internal/flags