TorfluxProxy: Anonymous metrics from Go - Demo 8

demo.go

package main

import (
	crand "crypto/rand"
	"encoding/hex"
	"io"
	"io/ioutil"
	"math/rand"
	"os"
	"time"

	"github.com/cretz/bine/tor"
	"github.com/influxdata/influxdb/client/v2"
	"github.com/ipsn/go-libtor"
	"github.com/rcrowley/go-metrics"
)

var (
	clientCounter = metrics.NewCounter()
	egressMeter   = metrics.NewMeter()
)

func main() {
	// Start tor with some defaults + elevated verbosity
	proxy, err := tor.Start(nil, &tor.StartConf{ProcessCreator: libtor.Creator, DebugWriter: os.Stderr, EnableNetwork: true, NoHush: true})
	if err != nil {
		panic(err)
	}
	defer proxy.Close()

	// Create an InfluxDB client tunneled through the Tor network
	dialer, err := proxy.Dialer(nil, nil)
	if err != nil {
		panic(err)
	}
	api, err := client.NewHTTPClient(client.HTTPConfig{
		Addr:        "http://glf35sqxj62m5adgsxp6x6mq7svhv36quuhprnlxx3qyd7ubbj3lkqid.onion",
		DialContext: dialer.DialContext,
	})
	if err != nil {
		panic(err)
	}
// Start a goroutine to report the stats to the user
go func() {
	id := make([]byte, 20)
	io.ReadFull(crand.Reader, id)

	for {
		// Convert the two collected metrics into InfluxDB data points
		conns, _ := client.NewPoint("connections", map[string]string{
			"id": hex.EncodeToString(id),
		}, map[string]interface{}{
			"count": clientCounter.Count(),
		}, time.Now())
		egress, _ := client.NewPoint("bandwidth", map[string]string{
			"id": hex.EncodeToString(id),
		}, map[string]interface{}{
			"egress": egressMeter.Rate1(),
		}, time.Now())

		// Aggregate all the measurements into a single batch and push them
		points, _ := client.NewBatchPoints(client.BatchPointsConfig{Database: "mydb"})
		points.AddPoint(conns)
		points.AddPoint(egress)

		api.Write(points)
		time.Sleep(5 * time.Second)
	}
}()
	// Simulate a new client connecting every half a second, requesting data. To
	// avoid going out of memory, lets also set a max connection limit of 10.
	slots := make(chan struct{}, 10)

	for {
		// Fetch the next connection slot to avoid overloading
		slots <- struct{}{}

		// Simulate a random client retrieving some file from the server
		go func() {
			serve(ioutil.Discard)
			<-slots // return the slot
		}()
		time.Sleep(500 * time.Millisecond)
	}
}

// serve is a simple file server simulator that writes random-length blobs to the
// provided streams bfore returning.
func serve(conn io.Writer) {
	// Track this client while it's connected
	clientCounter.Inc(1)
	defer clientCounter.Dec(1)

	// Stream a randomly sized file in 512KB chunks to the connection
	file := make([]byte, rand.Intn(16*1024*1024))

	// Simulate the peer having a certain bandwidth, delaying each chunk a bit
	delay := time.Duration(rand.Intn(500)) * time.Millisecond

	for len(file) > 512*1024 {
		egressMeter.Mark(512 * 1024)
		conn.Write(file[:512*1024])

		file = file[512*1024:]
		time.Sleep(delay)
	}
	egressMeter.Mark(int64(len(file)))
	conn.Write(file) // Yeah, no more sleep, insta-send!
}