wsjtx-exporter/cmd/alltxt2http/main.go

package main

import (
  "fmt"
  "github.com/jnovack/flag"
  log "github.com/sirupsen/logrus"
//  "strings"
//  "strconv"
  "time"
  "os"
  "bytes"
  "net/http"
  "io/ioutil"
  "encoding/json"
  "bufio"
//  "bufio"
//  "runtime"
//  "sync"
//  "github.com/mmcloughlin/geohash"
//  "github.com/tzneal/ham-go/dxcc"
//  "github.com/denzs/wsjtx-dashboards/shared/httpstuff"
  "github.com/hpcloud/tail"
  "github.com/prometheus/client_golang/prometheus/promhttp"
)

var alltxt      string
var station     string
var identifier  string
var password    string
var uri         string
var minlen      int
var readall     bool
var trace       bool

func usage() {
    fmt.Printf("Usage of %s:\n", os.Args[0])
    flag.PrintDefaults()
}

func init() {
  flag.StringVar(&alltxt, "in", "", "path to wsjt-x ALL.TXT")
  flag.StringVar(&station, "station", "YOURCALL", "your station/username")
  flag.StringVar(&identifier, "identifier", "unset", "your identifier like location of rigname")
  flag.StringVar(&password, "password", "", "your password")
  flag.StringVar(&uri, "uri", "http://http-exporter", "uri to your http-exporter")
  flag.IntVar(&minlen, "minlen", 16, "minimal length for line to be pushed")
  flag.BoolVar(&readall, "readall", false, "submit whole file if nothing found at server")
  flag.BoolVar(&trace, "trace", false, "log almost everything")
  flag.Parse()

  if alltxt == "" || station == "" || identifier == "" || uri == "" {
    usage()
    // on windows could do some fancy dialog magic to ask for parameters and service installation ;)
    os.Exit(1)
  }

  if trace {
    log.SetLevel(log.TraceLevel)
    log.Info("trace logging enabled")
  } else {
    log.Info("normal logging enabled")
  }

  formatter := &log.TextFormatter{
    FullTimestamp: true,
  }
  log.SetFormatter(formatter)

  log.Info("prometheus exporter enabled..")
  go func () {
    server := http.NewServeMux()
    server.Handle("/metrics", promhttp.Handler())
    http.ListenAndServe(":9123", server)
  } ()
}

func getTimestampFromDb() (int64, bool) {
  log.Trace("fetching last report from server")
	var ts LastTs

  client := http.Client{
		Timeout: time.Second * 2, // Timeout after 2 seconds
	}

	req, err := http.NewRequest(http.MethodGet, uri + "/timestamp/getLast?identifier=" + identifier, nil)
  req.SetBasicAuth(station, password)

	if err != nil {
    log.Errorf("error creating request: %s", err.Error())
	}

	req.Header.Set("X-Station", station)
	req.Header.Set("X-Identifier", identifier)

	res, getErr := client.Do(req)
	if getErr != nil {
    log.Errorf("error executing http request: %s", getErr.Error())
    os.Exit(1)
	}

  switch res.StatusCode {
    case 200:
      log.Trace("200 OK")
    case 401:
      log.Error("Authentication failed! Please check entered station and password..")
      os.Exit(1)
    case 404:
      log.Tracef("404 for %s:%s", station, identifier)
      return 0, false
    case 502:
      log.Errorf("Server reports internal problems (%d) please wait 30 seconds before trying again..", res.StatusCode)
      os.Exit(1)
    default:
      log.Panicf("uh well, there seems to be some serious shit going on: %d",res.StatusCode)
  }

//  if res.StatusCode != http.StatusOK {
//      if res.StatusCode == 404 {
//        // FIXME on windows show message to upload all.txt
//        log.Info("Yeah, this seems to be our first contact with the server, as it doesnt have any records for %s:%s", station, identifier)
//        return 0, false
//      }
//      fmt.Println("Non-OK HTTP status:", res.StatusCode)
//      log.Panic("uh well, there seems to be some serious shit going on...")
//  }

	if res.Body != nil {
		defer res.Body.Close()
	}

	body, readErr := ioutil.ReadAll(res.Body)
	if readErr != nil {
		log.Fatal(readErr)
	}

	jsonErr := json.Unmarshal(body, &ts)
	if jsonErr != nil {
		log.Fatal(jsonErr)
	}

  return ts.Last, true
}

// https://forum.golangbridge.org/t/how-to-find-the-offset-of-a-byte-in-a-large-binary-file/16457/2 ;)
// ScanFile returns the offset of the first occurence of a []byte in a file,
// or -1 if []byte was not found in file, and seeks to the beginning of the searched []byte
func findPosition(search []byte) (int64, bool) {
  f, err := os.Open(alltxt)
	if err != nil {
		log.Fatal(err)
	}

	ix := 0
	r := bufio.NewReader(f)
	offset := int64(0)
	for ix < len(search) {
		b, err := r.ReadByte()
		if err != nil {
			return 0, false
		}
		if search[ix] == b {
			ix++
		} else {
			ix = 0
		}
		offset++
	}
//	f.Seek(offset - int64(len(search)), 0 ) // Seeks to the beginning of the searched []byte
  
  //r.Close()
  //readseek.Close()
  f.Close()
	return offset - int64(len(search)), true
}

func unix2wsjtx(ts int64) (string) {
  t := time.Unix(ts, 0)
  s := t.UTC().Format("060102_150405")
  return s
}

func postLine(batchmode bool, lines chan string) {
  // FIXME implement batchmode

  for {
    select {
      case line := <- lines :
        l := Lines{}
        l.Text = append(l.Text, line)
        log.Tracef("line: %s", l.Text)

        jsonStr, err := json.Marshal(l)
        if err != nil {
          log.Errorf("error marshaling json: ", err.Error())
        }
        req, err := http.NewRequest("POST", uri + "/lines/insert", bytes.NewBuffer(jsonStr))
        if err != nil {
            log.Errorf("error creating request: %s", err.Error())
        }
        req.SetBasicAuth(station, password)
	      req.Header.Set("X-Station", station)
	      req.Header.Set("X-Identifier", identifier)
        req.Header.Set("Content-Type", "application/json")

        client := &http.Client{}
        resp, err := client.Do(req)
        if err != nil {
            log.Errorf("error executing http request: %s", err.Error())
        }

        // FIXME exit fatal if resposne != 200 OK
        log.Tracef("response: %s", resp.Status)
//        body, _ := ioutil.ReadAll(resp.Body)
//        log.Tracef("response: %s", string(body))
        resp.Body.Close()
        }
  }
  log.Trace("goodbye from postLine")
}

func readFile(offset int64, whence int, done chan bool) {
  workers := 128

  log.Tracef("start reading file")

  lines := make(chan string, workers)

  for i := 0; i < workers; i++ {
    go postLine((whence == 0), lines)
  }

  t, _:= tail.TailFile(alltxt, tail.Config{Follow: true, Location: &tail.SeekInfo{Offset: offset, Whence: whence},})
  log.Info("scanning for lines..")
  for line := range t.Lines {
    // FIXME
    // check for logline format!
    // at least minlen
    lines <- line.Text
  }

  close(lines)
  log.Trace("readfile is finished")
  done <- true
}

func main(){
  var lastDbTime  int64
  var foundts     bool
  var offset      int64
  var whence      int
  var foundoffset bool

  offset = 0

  if !readall {
    whence = 2
    lastDbTime, foundts = getTimestampFromDb()
    if foundts {
      log.Infof("last timestamp in db: %d", lastDbTime)
      wt := unix2wsjtx(lastDbTime)
      log.Infof("searching %s for %s",alltxt, wt)
      offset, foundoffset = findPosition([]byte(wt))
      if foundoffset {
        log.Infof("found %s at offset %d", wt, offset)
        whence = 0
      } else {
        log.Warnf("timestamp %s NOT found in %s! have you deleted your ALL.TXT or used an old identifier?", wt, alltxt)
        log.Warnf("sending spots from now on.. consider sending your %s to the database administrator for import", alltxt)
      }
    } else {
      log.Warnf("api has no data for %s:%s", station, identifier)
      log.Warnf("sending spots from now on.. consider sending your %s to the database administrator for import", alltxt)
    }
  } else {
    log.Info("readall mode enabled..")
    whence = 0
  }

  log.Tracef("offset: %d whence: %d", offset, whence)

  done := make(chan bool)
  go readFile(offset, whence, done)
  <- done

  log.Info("..done")
}