tailscale/logtail/example/logreprocess/logreprocess.go
Will Norris 71029cea2d all: update copyright and license headers
This updates all source files to use a new standard header for copyright
and license declaration.  Notably, copyright no longer includes a date,
and we now use the standard SPDX-License-Identifier header.

This commit was done almost entirely mechanically with perl, and then
some minimal manual fixes.

Updates #6865

Signed-off-by: Will Norris <will@tailscale.com>
2023-01-27 15:36:29 -08:00

116 lines
2.6 KiB
Go

// Copyright (c) Tailscale Inc & AUTHORS
// SPDX-License-Identifier: BSD-3-Clause
// The logreprocess program tails a log and reprocesses it.
package main
import (
"bufio"
"encoding/json"
"flag"
"io"
"log"
"net/http"
"os"
"strings"
"time"
"tailscale.com/logtail"
)
func main() {
collection := flag.String("c", "", "logtail collection name to read")
apiKey := flag.String("p", "", "logtail API key")
timeout := flag.Duration("t", 0, "timeout after which logreprocess quits")
flag.Parse()
if len(flag.Args()) != 0 {
flag.Usage()
os.Exit(1)
}
log.SetFlags(0)
if *timeout != 0 {
go func() {
<-time.After(*timeout)
log.Printf("logreprocess: timeout reached, quitting")
os.Exit(1)
}()
}
req, err := http.NewRequest("GET", "https://log.tailscale.io/c/"+*collection+"?stream=true", nil)
if err != nil {
log.Fatal(err)
}
req.SetBasicAuth(*apiKey, "")
resp, err := http.DefaultClient.Do(req)
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
b, err := io.ReadAll(resp.Body)
if err != nil {
log.Fatalf("logreprocess: read error %d: %v", resp.StatusCode, err)
}
log.Fatalf("logreprocess: read error %d: %s", resp.StatusCode, string(b))
}
tracebackCache := make(map[logtail.PublicID]*ProcessedMsg)
scanner := bufio.NewScanner(resp.Body)
for scanner.Scan() {
var msg Msg
if err := json.Unmarshal(scanner.Bytes(), &msg); err != nil {
log.Fatalf("logreprocess of %q: %v", string(scanner.Bytes()), err)
}
var pMsg *ProcessedMsg
if pMsg = tracebackCache[msg.Logtail.Instance]; pMsg != nil {
pMsg.Text += "\n" + msg.Text
if strings.HasPrefix(msg.Text, "Exception: ") {
delete(tracebackCache, msg.Logtail.Instance)
} else {
continue // write later
}
} else {
pMsg = &ProcessedMsg{
OrigInstance: msg.Logtail.Instance,
Text: msg.Text,
}
pMsg.Logtail.ClientTime = msg.Logtail.ClientTime
}
if strings.HasPrefix(msg.Text, "Traceback (most recent call last):") {
tracebackCache[msg.Logtail.Instance] = pMsg
continue // write later
}
b, err := json.Marshal(pMsg)
if err != nil {
log.Fatal(err)
}
log.Printf("%s", b)
}
if err := scanner.Err(); err != nil {
log.Fatal(err)
}
}
type Msg struct {
Logtail struct {
Instance logtail.PublicID `json:"instance"`
ClientTime time.Time `json:"client_time"`
} `json:"logtail"`
Text string `json:"text"`
}
type ProcessedMsg struct {
Logtail struct {
ClientTime time.Time `json:"client_time"`
} `json:"logtail"`
OrigInstance logtail.PublicID `json:"orig_instance"`
Text string `json:"text"`
}