Issue 1321 usage statistics (#1415)

* WIP

* Send usage analytics

* Improve loggin of cron tasks and fix test

* Implement appconfig method now that we are checking that as well

* Address review comments

changes/issue-1321-send-statistics

@@ -0,0 +1 @@
+* If enabled, it sends usage analytics to fleetdm.com. Fixes issue 1321

cmd/fleet/serve.go

@@ -1,15 +1,18 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"crypto/tls"
+	"encoding/json"
 	"fmt"
+	"net/url"
 
+	"github.com/e-dard/netbug"
 	"github.com/fleetdm/fleet/v4/server"
 
 	"io/ioutil"
 	"net/http"
-	"net/url"
 	"os"
 	"os/signal"
 	"regexp"
@@ -18,7 +21,6 @@ import (
 	"time"
 
 	"github.com/WatchBeam/clock"
-	"github.com/e-dard/netbug"
 	"github.com/fleetdm/fleet/v4/ee/server/licensing"
 	eeservice "github.com/fleetdm/fleet/v4/ee/server/service"
 	"github.com/fleetdm/fleet/v4/server/config"
@@ -202,7 +204,7 @@ the way that the Fleet server works.
 				}
 			}
 
-			cancelBackground := runCrons(ds)
+			cancelBackground := runCrons(ds, kitlog.With(logger, "component", "crons"))
 
 			// Flush seen hosts every second
 			go func() {
@@ -396,7 +398,38 @@ const (
 	LockKeyLeader = "leader"
 )
 
-func runCrons(ds fleet.Datastore) context.CancelFunc {
+func trySendStatistics(ds fleet.Datastore, frequency time.Duration, url string) error {
+	ac, err := ds.AppConfig()
+	if err != nil {
+		return err
+	}
+	if !ac.EnableAnalytics {
+		return nil
+	}
+
+	stats, shouldSend, err := ds.ShouldSendStatistics(frequency)
+	if err != nil {
+		return err
+	}
+	if !shouldSend {
+		return nil
+	}
+
+	statsBytes, err := json.Marshal(stats)
+	if err != nil {
+		return err
+	}
+	req, err := http.Post(url, "application/json", bytes.NewBuffer(statsBytes))
+	if err != nil {
+		return err
+	}
+	if req.StatusCode != http.StatusOK {
+		return errors.Errorf("Error posting to %s: %d", url, req.StatusCode)
+	}
+	return ds.RecordStatisticsSent()
+}
+
+func runCrons(ds fleet.Datastore, logger kitlog.Logger) context.CancelFunc {
 	locker, ok := ds.(Locker)
 	if !ok {
 		initFatal(errors.New("No global locker available"), "")
@@ -411,17 +444,36 @@ func runCrons(ds fleet.Datastore) context.CancelFunc {
 	go func() {
 		ticker := time.NewTicker(1 * time.Hour)
 		for {
+			level.Debug(logger).Log("waiting", "on ticker")
 			select {
 			case <-ticker.C:
+				level.Debug(logger).Log("waiting", "done")
 			case <-ctx.Done():
+				level.Debug(logger).Log("exit", "done with crons.")
 				break
 			}
 			if locked, err := locker.Lock(LockKeyLeader, ourIdentifier, time.Hour); err != nil || !locked {
+				level.Debug(logger).Log("leader", "Not the leader. Skipping...")
 				continue
 			}
-			ds.CleanupDistributedQueryCampaigns(time.Now())
-			ds.CleanupIncomingHosts(time.Now())
-			ds.CleanupCarves(time.Now())
+			_, err := ds.CleanupDistributedQueryCampaigns(time.Now())
+			if err != nil {
+				level.Error(logger).Log("err", "cleaning distributed query campaigns", "details", err)
+			}
+			err = ds.CleanupIncomingHosts(time.Now())
+			if err != nil {
+				level.Error(logger).Log("err", "cleaning incoming hosts", "details", err)
+			}
+			_, err = ds.CleanupCarves(time.Now())
+			if err != nil {
+				level.Error(logger).Log("err", "cleaning carves", "details", err)
+			}
+
+			err = trySendStatistics(ds, fleet.StatisticsFrequency, "https://fleetdm.com/api/v1/webhooks/receive-usage-analytics")
+			if err != nil {
+				level.Error(logger).Log("err", "sending statistics", "details", err)
+			}
+			level.Debug(logger).Log("loop", "done")
 		}
 	}()
 	return cancelBackground

cmd/fleet/serve_test.go

@@ -0,0 +1,97 @@
+package main
+
+import (
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/fleetdm/fleet/v4/server/fleet"
+	"github.com/fleetdm/fleet/v4/server/mock"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMaybeSendStatistics(t *testing.T) {
+	ds := new(mock.Store)
+
+	requestBody := ""
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requestBodyBytes, err := ioutil.ReadAll(r.Body)
+		require.NoError(t, err)
+		requestBody = string(requestBodyBytes)
+	}))
+	defer ts.Close()
+
+	ds.AppConfigFunc = func() (*fleet.AppConfig, error) {
+		return &fleet.AppConfig{EnableAnalytics: true}, nil
+	}
+
+	ds.ShouldSendStatisticsFunc = func(frequency time.Duration) (fleet.StatisticsPayload, bool, error) {
+		return fleet.StatisticsPayload{
+			AnonymousIdentifier: "ident",
+			FleetVersion:        "1.2.3",
+			NumHostsEnrolled:    999,
+		}, true, nil
+	}
+	recorded := false
+	ds.RecordStatisticsSentFunc = func() error {
+		recorded = true
+		return nil
+	}
+
+	err := trySendStatistics(ds, fleet.StatisticsFrequency, ts.URL)
+	require.NoError(t, err)
+	assert.True(t, recorded)
+	assert.Equal(t, `{"anonymousIdentifier":"ident","fleetVersion":"1.2.3","numHostsEnrolled":999}`, requestBody)
+}
+
+func TestMaybeSendStatisticsSkipsSendingIfNotNeeded(t *testing.T) {
+	ds := new(mock.Store)
+
+	called := false
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		called = true
+	}))
+	defer ts.Close()
+
+	ds.AppConfigFunc = func() (*fleet.AppConfig, error) {
+		return &fleet.AppConfig{EnableAnalytics: true}, nil
+	}
+
+	ds.ShouldSendStatisticsFunc = func(frequency time.Duration) (fleet.StatisticsPayload, bool, error) {
+		return fleet.StatisticsPayload{}, false, nil
+	}
+	recorded := false
+	ds.RecordStatisticsSentFunc = func() error {
+		recorded = true
+		return nil
+	}
+
+	err := trySendStatistics(ds, fleet.StatisticsFrequency, ts.URL)
+	require.NoError(t, err)
+	assert.False(t, recorded)
+	assert.False(t, called)
+}
+
+func TestMaybeSendStatisticsSkipsIfNotConfigured(t *testing.T) {
+	ds := new(mock.Store)
+
+	called := false
+
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		called = true
+	}))
+	defer ts.Close()
+
+	ds.AppConfigFunc = func() (*fleet.AppConfig, error) {
+		return &fleet.AppConfig{EnableAnalytics: false}, nil
+	}
+
+	err := trySendStatistics(ds, fleet.StatisticsFrequency, ts.URL)
+	require.NoError(t, err)
+	assert.False(t, called)
+}

server/datastore/mysql/hosts.go

@@ -310,6 +310,15 @@ func (d *Datastore) Host(id uint) (*fleet.Host, error) {
 	return host, nil
 }
 
+func (d *Datastore) amountEnrolledHosts() (int, error) {
+	var amount int
+	err := d.db.Get(&amount, `SELECT count(*) FROM hosts`)
+	if err != nil {
+		return 0, err
+	}
+	return amount, nil
+}
+
 func (d *Datastore) ListHosts(filter fleet.TeamFilter, opt fleet.HostListOptions) ([]*fleet.Host, error) {
 	sql := `SELECT
 		h.*,

server/datastore/mysql/migrations/tables/20210719153709_AddStatisticsTable.go

@@ -0,0 +1,32 @@
+package tables
+
+import (
+	"database/sql"
+
+	"github.com/pkg/errors"
+)
+
+func init() {
+	MigrationClient.AddMigration(Up_20210719153709, Down_20210719153709)
+}
+
+func Up_20210719153709(tx *sql.Tx) error {
+	sql := `
+		CREATE TABLE IF NOT EXISTS statistics (
+			id int(10) unsigned NOT NULL AUTO_INCREMENT,
+			created_at timestamp DEFAULT CURRENT_TIMESTAMP,
+			updated_at timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+			anonymous_identifier varchar(255) NOT NULL,
+			PRIMARY KEY (id) 
+		)
+	`
+	if _, err := tx.Exec(sql); err != nil {
+		return errors.Wrap(err, "create statistics")
+	}
+
+	return nil
+}
+
+func Down_20210719153709(tx *sql.Tx) error {
+	return nil
+}

server/datastore/mysql/statistics.go

@@ -0,0 +1,61 @@
+package mysql
+
+import (
+	"database/sql"
+	"time"
+
+	"github.com/fleetdm/fleet/v4/server"
+	"github.com/fleetdm/fleet/v4/server/fleet"
+	"github.com/kolide/kit/version"
+)
+
+type statistics struct {
+	fleet.UpdateCreateTimestamps
+	Identifier string `db:"anonymous_identifier"`
+}
+
+func (d *Datastore) ShouldSendStatistics(frequency time.Duration) (fleet.StatisticsPayload, bool, error) {
+	amountEnrolledHosts, err := d.amountEnrolledHosts()
+	if err != nil {
+		return fleet.StatisticsPayload{}, false, err
+	}
+
+	dest := statistics{}
+	err = d.db.Get(&dest, `SELECT created_at, updated_at, anonymous_identifier FROM statistics LIMIT 1`)
+	if err != nil {
+		if err == sql.ErrNoRows {
+			anonIdentifier, err := server.GenerateRandomText(64)
+			if err != nil {
+				return fleet.StatisticsPayload{}, false, err
+			}
+			_, err = d.db.Exec(`INSERT INTO statistics(anonymous_identifier) VALUES (?)`, anonIdentifier)
+			if err != nil {
+				return fleet.StatisticsPayload{}, false, err
+			}
+			return fleet.StatisticsPayload{
+				AnonymousIdentifier: anonIdentifier,
+				FleetVersion:        version.Version().Version,
+				NumHostsEnrolled:    amountEnrolledHosts,
+			}, true, nil
+		} else {
+			return fleet.StatisticsPayload{}, false, err
+		}
+	}
+	lastUpdated := dest.UpdatedAt
+	if dest.CreatedAt.After(dest.UpdatedAt) {
+		lastUpdated = dest.CreatedAt
+	}
+	if time.Now().Before(lastUpdated.Add(frequency)) {
+		return fleet.StatisticsPayload{}, false, nil
+	}
+	return fleet.StatisticsPayload{
+		AnonymousIdentifier: dest.Identifier,
+		FleetVersion:        version.Version().Version,
+		NumHostsEnrolled:    amountEnrolledHosts,
+	}, true, nil
+}
+
+func (d *Datastore) RecordStatisticsSent() error {
+	_, err := d.db.Exec(`UPDATE statistics SET updated_at = CURRENT_TIMESTAMP LIMIT 1`)
+	return err
+}

server/datastore/mysql/statistics_test.go

@@ -0,0 +1,66 @@
+package mysql
+
+import (
+	"testing"
+	"time"
+
+	"github.com/fleetdm/fleet/v4/server/fleet"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestShouldSendStatistics(t *testing.T) {
+	ds := CreateMySQLDS(t)
+	defer ds.Close()
+
+	_, err := ds.NewHost(&fleet.Host{
+		DetailUpdatedAt: time.Now(),
+		LabelUpdatedAt:  time.Now(),
+		SeenTime:        time.Now(),
+		NodeKey:         "1",
+		UUID:            "1",
+		Hostname:        "foo.local",
+		PrimaryIP:       "192.168.1.1",
+		PrimaryMac:      "30-65-EC-6F-C4-58",
+		OsqueryHostID:   "M",
+	})
+	require.NoError(t, err)
+
+	// First time running, we send statistics
+	stats, shouldSend, err := ds.ShouldSendStatistics(fleet.StatisticsFrequency)
+	require.NoError(t, err)
+	assert.True(t, shouldSend)
+	assert.NotEmpty(t, stats.AnonymousIdentifier)
+	assert.Equal(t, stats.NumHostsEnrolled, 1)
+	firstIdentifier := stats.AnonymousIdentifier
+
+	err = ds.RecordStatisticsSent()
+	require.NoError(t, err)
+
+	// If we try right away, it shouldn't ask to send
+	stats, shouldSend, err = ds.ShouldSendStatistics(fleet.StatisticsFrequency)
+	require.NoError(t, err)
+	assert.False(t, shouldSend)
+
+	time.Sleep(2)
+
+	_, err = ds.NewHost(&fleet.Host{
+		DetailUpdatedAt: time.Now(),
+		LabelUpdatedAt:  time.Now(),
+		SeenTime:        time.Now(),
+		NodeKey:         "2",
+		UUID:            "2",
+		Hostname:        "foo.local2",
+		PrimaryIP:       "192.168.1.2",
+		PrimaryMac:      "30-65-EC-6F-C4-59",
+		OsqueryHostID:   "S",
+	})
+	require.NoError(t, err)
+
+	// Lower the frequency to trigger an "outdated" sent
+	stats, shouldSend, err = ds.ShouldSendStatistics(time.Millisecond)
+	require.NoError(t, err)
+	assert.True(t, shouldSend)
+	assert.Equal(t, firstIdentifier, stats.AnonymousIdentifier)
+	assert.Equal(t, stats.NumHostsEnrolled, 2)
+}

server/fleet/datastore.go

@@ -18,6 +18,7 @@ type Datastore interface {
 	TeamStore
 	SoftwareStore
 	ActivitiesStore
+	StatisticsStore
 
 	Name() string
 	Drop() error

server/fleet/statistics.go

@@ -0,0 +1,18 @@
+package fleet
+
+import "time"
+
+type StatisticsPayload struct {
+	AnonymousIdentifier string `json:"anonymousIdentifier"`
+	FleetVersion        string `json:"fleetVersion"`
+	NumHostsEnrolled    int    `json:"numHostsEnrolled"`
+}
+
+type StatisticsStore interface {
+	ShouldSendStatistics(frequency time.Duration) (StatisticsPayload, bool, error)
+	RecordStatisticsSent() error
+}
+
+const (
+	StatisticsFrequency = time.Hour * 24 * 7
+)

server/mock/datastore.go

@@ -38,6 +38,7 @@ type Store struct {
 	CarveStore
 	SoftwareStore
 	ActivitiesStore
+	StatisticsStore
 }
 
 func (m *Store) Drop() error {

server/mock/datastore_statistics.go

@@ -0,0 +1,32 @@
+// Automatically generated by mockimpl. DO NOT EDIT!
+
+package mock
+
+import (
+	"time"
+
+	"github.com/fleetdm/fleet/v4/server/fleet"
+)
+
+var _ fleet.StatisticsStore = (*StatisticsStore)(nil)
+
+type ShouldSendStatisticsFunc func(frequency time.Duration) (fleet.StatisticsPayload, bool, error)
+type RecordStatisticsSentFunc func() error
+
+type StatisticsStore struct {
+	ShouldSendStatisticsFunc        ShouldSendStatisticsFunc
+	ShouldSendStatisticsFuncInvoked bool
+
+	RecordStatisticsSentFunc        RecordStatisticsSentFunc
+	RecordStatisticsSentFuncInvoked bool
+}
+
+func (s *StatisticsStore) ShouldSendStatistics(frequency time.Duration) (fleet.StatisticsPayload, bool, error) {
+	s.ShouldSendStatisticsFuncInvoked = true
+	return s.ShouldSendStatisticsFunc(frequency)
+}
+
+func (s *StatisticsStore) RecordStatisticsSent() error {
+	s.RecordStatisticsSentFuncInvoked = true
+	return s.RecordStatisticsSentFunc()
+}