Discharge7443/fleet
1
0
Fork 0
Code Issues Pull Requests Actions 5 Packages Projects Releases Wiki Activity

Refactor label membership storage

Browse Source 
Label membership is now stored in the label_membership table. This is
done in preparation for adding "manual" labels, as previously label
membership was associated directly with label query executions.

Label queries are now all executed at the same time, rather than on
separate intervals. This simplifies the calculation of which distributed
queries need to be run when a host checks in.
This commit is contained in:
Zachary Wasserman
2020-07-21 14:05:46 -07:00
committed by Zachary Wasserman
parent b0595748f1
commit 608772917c
20 changed files with 305 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/datastore/datastore_campaigns_test.go
+3 -6
View File
@@ -88,10 +88,9 @@ func testCleanupDistributedQueryCampaigns(t *testing.T, ds kolide.Datastore) {
// Cleanup and verify that nothing changed (because time has not
// advanced)
expired, deleted, err := ds.CleanupDistributedQueryCampaigns(mockClock.Now())
expired, err := ds.CleanupDistributedQueryCampaigns(mockClock.Now())
require.Nil(t, err)
assert.Equal(t, uint(0), expired)
assert.Equal(t, uint(0), deleted)
{
retrieved, err := ds.DistributedQueryCampaign(c1.ID)
@@ -112,10 +111,9 @@ func testCleanupDistributedQueryCampaigns(t *testing.T, ds kolide.Datastore) {
// Cleanup and verify that the campaign was expired and executions
// deleted appropriately
expired, deleted, err = ds.CleanupDistributedQueryCampaigns(mockClock.Now())
expired, err = ds.CleanupDistributedQueryCampaigns(mockClock.Now())
require.Nil(t, err)
assert.Equal(t, uint(1), expired)
assert.Equal(t, uint(2), deleted)
{
// c1 should now be complete
retrieved, err := ds.DistributedQueryCampaign(c1.ID)
@@ -134,10 +132,9 @@ func testCleanupDistributedQueryCampaigns(t *testing.T, ds kolide.Datastore) {
// Cleanup and verify that the campaign was expired and executions
// deleted appropriately
expired, deleted, err = ds.CleanupDistributedQueryCampaigns(mockClock.Now())
expired, err = ds.CleanupDistributedQueryCampaigns(mockClock.Now())
require.Nil(t, err)
assert.Equal(t, uint(1), expired)
assert.Equal(t, uint(3), deleted)
{
retrieved, err := ds.DistributedQueryCampaign(c1.ID)
require.Nil(t, err)
server/datastore/datastore_hosts_test.go
+20
View File
@@ -10,6 +10,7 @@ import (
"github.com/WatchBeam/clock"
"github.com/kolide/fleet/server/kolide"
"github.com/kolide/fleet/server/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -42,6 +43,7 @@ var enrollTests = []struct {
func testSaveHosts(t *testing.T, ds kolide.Datastore) {
host, err := ds.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
NodeKey: "1",
UUID: "1",
@@ -92,6 +94,7 @@ func testSaveHosts(t *testing.T, ds kolide.Datastore) {
func testDeleteHost(t *testing.T, ds kolide.Datastore) {
host, err := ds.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
NodeKey: "1",
UUID: "1",
@@ -110,6 +113,7 @@ func testDeleteHost(t *testing.T, ds kolide.Datastore) {
func testIdempotentDeleteHost(t *testing.T, ds kolide.Datastore) {
host, err := ds.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
NodeKey: "1",
UUID: "1",
@@ -133,6 +137,7 @@ func testListHost(t *testing.T, ds kolide.Datastore) {
for i := 0; i < 10; i++ {
host, err := ds.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
OsqueryHostID: strconv.Itoa(i),
NodeKey: fmt.Sprintf("%d", i),
@@ -173,6 +178,7 @@ func testListHost(t *testing.T, ds kolide.Datastore) {
}
func testEnrollHost(t *testing.T, ds kolide.Datastore) {
test.AddAllHostsLabel(t, ds)
var hosts []*kolide.Host
for _, tt := range enrollTests {
h, err := ds.EnrollHost(tt.uuid, tt.nodeKey, "default")
@@ -185,6 +191,7 @@ func testEnrollHost(t *testing.T, ds kolide.Datastore) {
}
func testAuthenticateHost(t *testing.T, ds kolide.Datastore) {
test.AddAllHostsLabel(t, ds)
for _, tt := range enrollTests {
h, err := ds.EnrollHost(tt.uuid, tt.nodeKey, "default")
require.Nil(t, err)
@@ -205,6 +212,7 @@ func testSearchHosts(t *testing.T, ds kolide.Datastore) {
_, err := ds.NewHost(&kolide.Host{
OsqueryHostID: "1234",
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
NodeKey: "1",
UUID: "1",
@@ -215,6 +223,7 @@ func testSearchHosts(t *testing.T, ds kolide.Datastore) {
h2, err := ds.NewHost(&kolide.Host{
OsqueryHostID: "5679",
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
NodeKey: "2",
UUID: "2",
@@ -225,6 +234,7 @@ func testSearchHosts(t *testing.T, ds kolide.Datastore) {
h3, err := ds.NewHost(&kolide.Host{
OsqueryHostID: "99999",
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
NodeKey: "3",
UUID: "abc-def-ghi",
@@ -288,6 +298,7 @@ func testSearchHostsLimit(t *testing.T, ds kolide.Datastore) {
for i := 0; i < 15; i++ {
_, err := ds.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
OsqueryHostID: fmt.Sprintf("host%d", i),
NodeKey: fmt.Sprintf("%d", i),
@@ -323,6 +334,7 @@ func testGenerateHostStatusStatistics(t *testing.T, ds kolide.Datastore) {
OsqueryHostID: "1",
NodeKey: "1",
DetailUpdateTime: mockClock.Now().Add(-30 * time.Second),
LabelUpdateTime: mockClock.Now().Add(-30 * time.Second),
SeenTime: mockClock.Now().Add(-30 * time.Second),
})
require.Nil(t, err)
@@ -336,6 +348,7 @@ func testGenerateHostStatusStatistics(t *testing.T, ds kolide.Datastore) {
OsqueryHostID: "2",
NodeKey: "2",
DetailUpdateTime: mockClock.Now().Add(-1 * time.Minute),
LabelUpdateTime: mockClock.Now().Add(-1 * time.Minute),
SeenTime: mockClock.Now().Add(-1 * time.Minute),
})
require.Nil(t, err)
@@ -349,6 +362,7 @@ func testGenerateHostStatusStatistics(t *testing.T, ds kolide.Datastore) {
OsqueryHostID: "3",
NodeKey: "3",
DetailUpdateTime: mockClock.Now().Add(-1 * time.Hour),
LabelUpdateTime: mockClock.Now().Add(-1 * time.Hour),
SeenTime: mockClock.Now().Add(-1 * time.Hour),
})
require.Nil(t, err)
@@ -362,6 +376,7 @@ func testGenerateHostStatusStatistics(t *testing.T, ds kolide.Datastore) {
OsqueryHostID: "4",
NodeKey: "4",
DetailUpdateTime: mockClock.Now().Add(-35 * (24 * time.Hour)),
LabelUpdateTime: mockClock.Now().Add(-35 * (24 * time.Hour)),
SeenTime: mockClock.Now().Add(-35 * (24 * time.Hour)),
})
require.Nil(t, err)
@@ -393,6 +408,7 @@ func testMarkHostSeen(t *testing.T, ds kolide.Datastore) {
UUID: "1",
NodeKey: "1",
DetailUpdateTime: aDayAgo,
LabelUpdateTime: aDayAgo,
SeenTime: aDayAgo,
})
assert.Nil(t, err)
@@ -424,6 +440,7 @@ func testCleanupIncomingHosts(t *testing.T, ds kolide.Datastore) {
UUID: "1",
NodeKey: "1",
DetailUpdateTime: mockClock.Now(),
LabelUpdateTime: mockClock.Now(),
SeenTime: mockClock.Now(),
})
require.Nil(t, err)
@@ -436,6 +453,7 @@ func testCleanupIncomingHosts(t *testing.T, ds kolide.Datastore) {
HostName: "foobar",
OsqueryVersion: "3.2.3",
DetailUpdateTime: mockClock.Now(),
LabelUpdateTime: mockClock.Now(),
SeenTime: mockClock.Now(),
})
require.Nil(t, err)
@@ -463,6 +481,7 @@ func testHostIDsByName(t *testing.T, ds kolide.Datastore) {
for i := 0; i < 10; i++ {
_, err := ds.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
OsqueryHostID: fmt.Sprintf("host%d", i),
NodeKey: fmt.Sprintf("%d", i),
@@ -482,6 +501,7 @@ func testHostAdditional(t *testing.T, ds kolide.Datastore) {
_, err := ds.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
SeenTime: time.Now(),
LabelUpdateTime: time.Now(),
OsqueryHostID: "foobar",
NodeKey: "nodekey",
UUID: "uuid",
server/datastore/datastore_labels_test.go
+63 -42
View File
@@ -8,11 +8,13 @@ import (
"time"
"github.com/kolide/fleet/server/kolide"
"github.com/kolide/fleet/server/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func testLabels(t *testing.T, db kolide.Datastore) {
test.AddAllHostsLabel(t, db)
hosts := []kolide.Host{}
var host *kolide.Host
var err error
@@ -21,45 +23,38 @@ func testLabels(t *testing.T, db kolide.Datastore) {
require.Nil(t, err, "enrollment should succeed")
hosts = append(hosts, *host)
}
host.Platform = "darwin"
require.NoError(t, db.SaveHost(host))
baseTime := time.Now()
// No queries should be returned before labels or queries added
// Only 'All Hosts' query should be returned
queries, err := db.LabelQueriesForHost(host, baseTime)
assert.Nil(t, err)
assert.Empty(t, queries)
assert.Len(t, queries, 1)
// No labels should match
// Only 'All Hosts' label should be returned
labels, err := db.ListLabelsForHost(host.ID)
assert.Nil(t, err)
assert.Empty(t, labels)
// No queries should be returned before labels added
queries, err = db.LabelQueriesForHost(host, baseTime)
assert.Nil(t, err)
assert.Empty(t, queries)
assert.Len(t, labels, 1)
newLabels := []*kolide.LabelSpec{
// Note these are intentionally out of order
&kolide.LabelSpec{
ID: 1,
Name: "label3",
Query: "query3",
Platform: "darwin",
},
&kolide.LabelSpec{
ID: 2,
Name: "label1",
Query: "query1",
},
&kolide.LabelSpec{
ID: 3,
Name: "label2",
Query: "query2",
Platform: "darwin",
},
&kolide.LabelSpec{
ID: 4,
Name: "label4",
Query: "query4",
Platform: "darwin",
@@ -69,10 +64,11 @@ func testLabels(t *testing.T, db kolide.Datastore) {
require.Nil(t, err)
expectQueries := map[string]string{
"1": "query3",
"2": "query1",
"3": "query2",
"4": "query4",
"1": "select 1",
"2": "query3",
"3": "query1",
"4": "query2",
"5": "query4",
}
host.Platform = "darwin"
@@ -85,33 +81,54 @@ func testLabels(t *testing.T, db kolide.Datastore) {
// No labels should match with no results yet
labels, err = db.ListLabelsForHost(host.ID)
assert.Nil(t, err)
assert.Empty(t, labels)
assert.Len(t, labels, 1)
// Record a query execution
err = db.RecordLabelQueryExecutions(host, map[uint]bool{1: true}, baseTime)
err = db.RecordLabelQueryExecutions(host, map[uint]bool{1: true, 2: false, 3: true, 4: false, 5: false}, baseTime)
assert.Nil(t, err)
// Use a 10 minute interval, so the query we just added should show up
queries, err = db.LabelQueriesForHost(host, time.Now().Add(-(10 * time.Minute)))
assert.Nil(t, err)
delete(expectQueries, "1")
assert.Equal(t, expectQueries, queries)
host, err = db.Host(host.ID)
require.NoError(t, err)
host.LabelUpdateTime = baseTime
// Record an old query execution -- Shouldn't change the return
err = db.RecordLabelQueryExecutions(host, map[uint]bool{2: true}, baseTime.Add(-1*time.Hour))
// Now no queries should be returned
queries, err = db.LabelQueriesForHost(host, baseTime.Add(-1*time.Minute))
assert.Nil(t, err)
queries, err = db.LabelQueriesForHost(host, time.Now().Add(-(10 * time.Minute)))
assert.Nil(t, err)
assert.Equal(t, expectQueries, queries)
assert.Len(t, queries, 0)
// Record a newer execution for that query and another
err = db.RecordLabelQueryExecutions(host, map[uint]bool{2: false, 3: true}, baseTime)
assert.Nil(t, err)
// Ensure enough gap in created_at
time.Sleep(2 * time.Second)
// Now these should no longer show up in the necessary to run queries
delete(expectQueries, "2")
delete(expectQueries, "3")
queries, err = db.LabelQueriesForHost(host, time.Now().Add(-(10 * time.Minute)))
// A new label targeting another platform should not effect the labels for
// this host
err = db.ApplyLabelSpecs([]*kolide.LabelSpec{
&kolide.LabelSpec{
Name: "label5",
Platform: "not-matching",
Query: "query5",
},
})
require.NoError(t, err)
queries, err = db.LabelQueriesForHost(host, baseTime.Add(-1*time.Minute))
assert.Nil(t, err)
assert.Len(t, queries, 0)
// If a new label is added, all labels should be returned
err = db.ApplyLabelSpecs([]*kolide.LabelSpec{
&kolide.LabelSpec{
Name: "label6",
Platform: "",
Query: "query6",
},
})
require.NoError(t, err)
expectQueries["7"] = "query6"
queries, err = db.LabelQueriesForHost(host, baseTime.Add(-1*time.Minute))
assert.Nil(t, err)
assert.Len(t, queries, 6)
// After expiration, all queries should be returned
queries, err = db.LabelQueriesForHost(host, baseTime.Add((2 * time.Minute)))
assert.Nil(t, err)
assert.Equal(t, expectQueries, queries)
@@ -121,8 +138,8 @@ func testLabels(t *testing.T, db kolide.Datastore) {
if assert.Len(t, labels, 2) {
labelNames := []string{labels[0].Name, labels[1].Name}
sort.Strings(labelNames)
assert.Equal(t, "label2", labelNames[0])
assert.Equal(t, "label3", labelNames[1])
assert.Equal(t, "All Hosts", labelNames[0])
assert.Equal(t, "label1", labelNames[1])
}
// A host that hasn't executed any label queries should still be asked
@@ -130,13 +147,13 @@ func testLabels(t *testing.T, db kolide.Datastore) {
hosts[0].Platform = "darwin"
queries, err = db.LabelQueriesForHost(&hosts[0], time.Now())
assert.Nil(t, err)
assert.Len(t, queries, 4)
assert.Len(t, queries, 6)
// There should still be no labels returned for a host that never
// executed any label queries
// Only the 'All Hosts' label should apply for a host with no labels
// executed.
labels, err = db.ListLabelsForHost(hosts[0].ID)
assert.Nil(t, err)
assert.Empty(t, labels)
assert.Len(t, labels, 1)
}
func testManagingLabelsOnPacks(t *testing.T, ds kolide.Datastore) {
@@ -273,6 +290,7 @@ func testSearchLabelsLimit(t *testing.T, db kolide.Datastore) {
func testListHostsInLabel(t *testing.T, db kolide.Datastore) {
h1, err := db.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
OsqueryHostID: "1",
NodeKey: "1",
@@ -283,6 +301,7 @@ func testListHostsInLabel(t *testing.T, db kolide.Datastore) {
h2, err := db.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
OsqueryHostID: "2",
NodeKey: "2",
@@ -293,6 +312,7 @@ func testListHostsInLabel(t *testing.T, db kolide.Datastore) {
h3, err := db.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
OsqueryHostID: "3",
NodeKey: "3",
@@ -344,6 +364,7 @@ func testListUniqueHostsInLabels(t *testing.T, db kolide.Datastore) {
for i := 0; i < 4; i++ {
h, err := db.NewHost(&kolide.Host{
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
OsqueryHostID: strconv.Itoa(i),
NodeKey: strconv.Itoa(i),
server/datastore/datastore_queries_test.go
+2
View File
@@ -12,6 +12,8 @@ import (
)
func testApplyQueries(t *testing.T, ds kolide.Datastore) {
test.AddAllHostsLabel(t, ds)
zwass := test.NewUser(t, ds, "Zach", "zwass", "zwass@kolide.co", true)
groob := test.NewUser(t, ds, "Victor", "groob", "victor@kolide.co", true)
expectedQueries := []*kolide.Query{
server/datastore/datastore_targets_test.go
+5
View File
@@ -7,6 +7,7 @@ import (
"github.com/WatchBeam/clock"
"github.com/kolide/fleet/server/kolide"
"github.com/kolide/fleet/server/test"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -24,6 +25,7 @@ func testCountHostsInTargets(t *testing.T, ds kolide.Datastore) {
h, err := ds.NewHost(&kolide.Host{
OsqueryHostID: strconv.Itoa(hostCount),
DetailUpdateTime: mockClock.Now(),
LabelUpdateTime: mockClock.Now(),
SeenTime: mockClock.Now(),
NodeKey: strconv.Itoa(hostCount),
DistributedInterval: distributedInterval,
@@ -119,6 +121,8 @@ func testCountHostsInTargets(t *testing.T, ds kolide.Datastore) {
}
func testHostStatus(t *testing.T, ds kolide.Datastore) {
test.AddAllHostsLabel(t, ds)
if ds.Name() == "inmem" {
t.Skip("inmem is being deprecated, test skipped")
}
@@ -189,6 +193,7 @@ func testHostIDsInTargets(t *testing.T, ds kolide.Datastore) {
OsqueryHostID: strconv.Itoa(hostCount),
NodeKey: strconv.Itoa(hostCount),
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
})
require.Nil(t, err)
server/datastore/datastore_unicode_test.go
+1
View File
@@ -30,6 +30,7 @@ func testUnicode(t *testing.T, ds kolide.Datastore) {
host, err := ds.NewHost(&kolide.Host{
HostName: "🍌",
DetailUpdateTime: time.Now(),
LabelUpdateTime: time.Now(),
SeenTime: time.Now(),
})
require.Nil(t, err)
server/datastore/inmem/campaigns.go
+2 -2
View File
@@ -72,7 +72,7 @@ func (d *Datastore) NewDistributedQueryCampaignTarget(target *kolide.Distributed
return target, nil
}
func (d *Datastore) CleanupDistributedQueryCampaigns(now time.Time) (expired uint, deleted uint, err error) {
func (d *Datastore) CleanupDistributedQueryCampaigns(now time.Time) (expired uint, err error) {
d.mtx.Lock()
defer d.mtx.Unlock()
@@ -86,5 +86,5 @@ func (d *Datastore) CleanupDistributedQueryCampaigns(now time.Time) (expired uin
}
}
return expired, 0, nil
return expired, nil
}
server/datastore/mysql/campaigns.go
+5 -25
View File
@@ -108,8 +108,8 @@ func (d *Datastore) NewDistributedQueryCampaignTarget(target *kolide.Distributed
return target, nil
}
func (d *Datastore) CleanupDistributedQueryCampaigns(now time.Time) (expired uint, deleted uint, err error) {
// First expire old waiting and running campaigns
func (d *Datastore) CleanupDistributedQueryCampaigns(now time.Time) (expired uint, err error) {
// Expire old waiting/running campaigns
sqlStatement := `
UPDATE distributed_query_campaigns
SET status = ?
@@ -120,33 +120,13 @@ func (d *Datastore) CleanupDistributedQueryCampaigns(now time.Time) (expired uin
kolide.QueryWaiting, now.Add(-1*time.Minute),
kolide.QueryRunning, now.Add(-24*time.Hour))
if err != nil {
return expired, deleted, errors.Wrap(err, "updating distributed query campaign")
return 0, errors.Wrap(err, "updating distributed query campaign")
}
exp, err := result.RowsAffected()
if err != nil {
return expired, deleted, errors.Wrap(err, "rows effected updating distributed query campaign")
}
expired = uint(exp)
// Now delete executions for expired campaigns
sqlStatement = `
DELETE dqe
FROM distributed_query_executions dqe
JOIN distributed_query_campaigns dqc
ON dqe.distributed_query_campaign_id = dqc.id
WHERE dqc.status = ?
`
result, err = d.db.Exec(sqlStatement, kolide.QueryComplete)
if err != nil {
return expired, deleted, errors.Wrap(err, "deleting distributed campaign executions")
return 0, errors.Wrap(err, "rows effected updating distributed query campaign")
}
del, err := result.RowsAffected()
if err != nil {
return expired, deleted, errors.Wrap(err, "rows effected deleting distributed campaign")
}
deleted = uint(del)
return expired, deleted, nil
return uint(exp), nil
}
server/datastore/mysql/hosts.go
+28 -52
View File
@@ -15,6 +15,7 @@ func (d *Datastore) NewHost(host *kolide.Host) (*kolide.Host, error) {
INSERT INTO hosts (
osquery_host_id,
detail_update_time,
label_update_time,
node_key,
host_name,
uuid,
@@ -25,11 +26,23 @@ func (d *Datastore) NewHost(host *kolide.Host) (*kolide.Host, error) {
physical_memory,
seen_time
)
VALUES( ?,?,?,?,?,?,?,?,?,?,? )
VALUES( ?,?,?,?,?,?,?,?,?,?,?,? )
`
result, err := d.db.Exec(sqlStatement, host.OsqueryHostID, host.DetailUpdateTime,
host.NodeKey, host.HostName, host.UUID, host.Platform, host.OsqueryVersion,
host.OSVersion, host.Uptime, host.PhysicalMemory, host.SeenTime)
result, err := d.db.Exec(
sqlStatement,
host.OsqueryHostID,
host.DetailUpdateTime,
host.LabelUpdateTime,
host.NodeKey,
host.HostName,
host.UUID,
host.Platform,
host.OsqueryVersion,
host.OSVersion,
host.Uptime,
host.PhysicalMemory,
host.SeenTime,
)
if err != nil {
return nil, errors.Wrap(err, "new host")
}
@@ -43,6 +56,7 @@ func (d *Datastore) SaveHost(host *kolide.Host) error {
sqlStatement := `
UPDATE hosts SET
detail_update_time = ?,
label_update_time = ?,
node_key = ?,
host_name = ?,
uuid = ?,
@@ -76,6 +90,7 @@ func (d *Datastore) SaveHost(host *kolide.Host) error {
`
_, err := d.db.Exec(sqlStatement,
host.DetailUpdateTime,
host.LabelUpdateTime,
host.NodeKey,
host.HostName,
host.UUID,
@@ -218,23 +233,23 @@ func (d *Datastore) EnrollHost(osqueryHostID, nodeKey, secretName string) (*koli
return nil, fmt.Errorf("missing osquery host identifier")
}
detailUpdateTime := time.Unix(0, 0).Add(24 * time.Hour)
zeroTime := time.Unix(0, 0).Add(24 * time.Hour)
sqlInsert := `
INSERT INTO hosts (
detail_update_time,
label_update_time,
osquery_host_id,
seen_time,
node_key,
enroll_secret_name
) VALUES (?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
node_key = VALUES(node_key),
deleted = FALSE
`
var result sql.Result
result, err := d.db.Exec(sqlInsert, detailUpdateTime, osqueryHostID, time.Now().UTC(), nodeKey, secretName)
result, err := d.db.Exec(sqlInsert, zeroTime, zeroTime, osqueryHostID, time.Now().UTC(), nodeKey, secretName)
if err != nil {
return nil, errors.Wrap(err, "inserting")
@@ -250,6 +265,11 @@ func (d *Datastore) EnrollHost(osqueryHostID, nodeKey, secretName string) (*koli
return nil, errors.Wrap(err, "getting the host to return")
}
_, err = d.db.Exec(`INSERT INTO label_membership (host_id, label_id) VALUES (?, (SELECT id FROM labels WHERE name = 'All Hosts' AND label_type = 1))`, id)
if err != nil {
return nil, errors.Wrap(err, "insert new host into all hosts label")
}
return host, nil
}
@@ -437,50 +457,6 @@ func (d *Datastore) SearchHosts(query string, omit ...uint) ([]*kolide.Host, err
}
func (d *Datastore) DistributedQueriesForHost(host *kolide.Host) (map[uint]string, error) {
sqlStatement := `
SELECT DISTINCT dqc.id, q.query
FROM distributed_query_campaigns dqc
JOIN distributed_query_campaign_targets dqct
ON (dqc.id = dqct.distributed_query_campaign_id)
LEFT JOIN label_query_executions lqe
ON (dqct.type = ? AND dqct.target_id = lqe.label_id AND lqe.matches)
LEFT JOIN hosts h
ON ((dqct.type = ? AND lqe.host_id = h.id) OR (dqct.type = ? AND dqct.target_id = h.id))
LEFT JOIN distributed_query_executions dqe
ON (h.id = dqe.host_id AND dqc.id = dqe.distributed_query_campaign_id)
JOIN queries q
ON (dqc.query_id = q.id)
WHERE dqe.status IS NULL AND dqc.status = ? AND h.id = ?
AND NOT q.deleted
AND NOT dqc.deleted
`
rows, err := d.db.Query(sqlStatement, kolide.TargetLabel, kolide.TargetLabel,
kolide.TargetHost, kolide.QueryRunning, host.ID)
if err != nil {
return nil, errors.Wrap(err, "finding distributed queries for host")
}
defer rows.Close()
results := map[uint]string{}
for rows.Next() {
var (
id uint
query string
)
err = rows.Scan(&id, &query)
if err != nil {
return nil, errors.Wrap(err, "scanning query results")
}
results[id] = query
}
return results, nil
}
func (d *Datastore) HostIDsByName(hostnames []string) ([]uint, error) {
if len(hostnames) == 0 {
return []uint{}, nil
server/datastore/mysql/labels.go
+73 -49
View File
@@ -2,6 +2,7 @@ package mysql
import (
"database/sql"
"strings"
"time"
"github.com/jmoiron/sqlx"
@@ -177,21 +178,27 @@ func (d *Datastore) ListLabels(opt kolide.ListOptions) ([]*kolide.Label, error)
}
func (d *Datastore) LabelQueriesForHost(host *kolide.Host, cutoff time.Time) (map[string]string, error) {
sqlStatment := `
SELECT l.id, l.query
FROM labels l
WHERE (l.platform = ? OR l.platform = '')
AND NOT l.deleted
AND l.id NOT IN /* subtract the set of executions that are recent enough */
(
SELECT l.id
FROM labels l
JOIN label_query_executions lqe
ON lqe.label_id = l.id
WHERE lqe.host_id = ? AND lqe.updated_at > ?
)
`
rows, err := d.db.Query(sqlStatment, host.Platform, host.ID, cutoff)
var rows *sql.Rows
var err error
if host.LabelUpdateTime.Before(cutoff) {
// Retrieve all labels (with matching platform) for this host
sql := `
SELECT id, query
FROM labels
WHERE platform = ? OR platform = ''`
rows, err = d.db.Query(sql, host.Platform)
} else {
// Retrieve all labels (with matching platform) iff there is a label
// that has been created since this host last reported label query
// executions
sql := `
SELECT id, query
FROM labels
WHERE ((SELECT max(created_at) FROM labels WHERE platform = ? OR platform = '') > ?)
AND (platform = ? OR platform = '')`
rows, err = d.db.Query(sql, host.Platform, host.LabelUpdateTime, host.Platform)
}
if err != nil && err != sql.ErrNoRows {
return nil, errors.Wrap(err, "selecting label queries for host")
}
@@ -214,31 +221,54 @@ func (d *Datastore) LabelQueriesForHost(host *kolide.Host, cutoff time.Time) (ma
}
func (d *Datastore) RecordLabelQueryExecutions(host *kolide.Host, results map[uint]bool, updated time.Time) error {
sqlStatement := `
INSERT INTO label_query_executions (updated_at, matches, label_id, host_id) VALUES
`
// Loop through results, collecting which labels we need to insert/update,
// and which we need to delete
vals := []interface{}{}
bindvars := ""
bindvars := []string{}
removes := []uint{}
for labelID, result := range results {
if bindvars != "" {
bindvars += ","
for labelID, matches := range results {
if matches {
// Add/update row
bindvars = append(bindvars, "(?,?,?)")
vals = append(vals, updated, labelID, host.ID)
} else {
// Delete row
removes = append(removes, labelID)
}
bindvars += "(?,?,?,?)"
vals = append(vals, updated, result, labelID, host.ID)
}
sqlStatement += bindvars
sqlStatement += `
ON DUPLICATE KEY UPDATE
updated_at = VALUES(updated_at),
matches = VALUES(matches)
`
// Complete inserts if necessary
if len(vals) > 0 {
sql := `
INSERT INTO label_membership (updated_at, label_id, host_id) VALUES
`
sql += strings.Join(bindvars, ",") +
`
ON DUPLICATE KEY UPDATE
updated_at = VALUES(updated_at)
`
_, err := d.db.Exec(sqlStatement, vals...)
if err != nil {
return errors.Wrap(err, "inserting label query execution")
_, err := d.db.Exec(sql, vals...)
if err != nil {
return errors.Wrap(err, "insert label query executions")
}
}
// Complete deletions if necessary
if len(removes) > 0 {
sql := `
DELETE FROM label_membership WHERE host_id = ? AND label_id IN (?)
`
query, args, err := sqlx.In(sql, host.ID, removes)
if err != nil {
return errors.Wrap(err, "IN for DELETE FROM label_membership")
}
query = d.db.Rebind(query)
_, err = d.db.Exec(query, args...)
if err != nil {
return errors.Wrap(err, "delete label query executions")
}
}
return nil
@@ -247,11 +277,9 @@ func (d *Datastore) RecordLabelQueryExecutions(host *kolide.Host, results map[ui
// ListLabelsForHost returns a list of kolide.Label for a given host id.
func (d *Datastore) ListLabelsForHost(hid uint) ([]kolide.Label, error) {
sqlStatement := `
SELECT labels.* from labels, label_query_executions lqe
WHERE lqe.host_id = ?
AND lqe.label_id = labels.id
AND lqe.matches
AND NOT labels.deleted
SELECT labels.* from labels JOIN label_membership lm
WHERE lm.host_id = ?
AND lm.label_id = labels.id
`
labels := []kolide.Label{}
@@ -269,12 +297,10 @@ func (d *Datastore) ListLabelsForHost(hid uint) ([]kolide.Label, error) {
func (d *Datastore) ListHostsInLabel(lid uint, opt kolide.ListOptions) ([]kolide.Host, error) {
sql := `
SELECT h.*
FROM label_query_executions lqe
FROM label_membership lm
JOIN hosts h
ON lqe.host_id = h.id
WHERE lqe.label_id = ?
AND lqe.matches = 1
AND NOT h.deleted
ON lm.host_id = h.id
WHERE lm.label_id = ?
`
sql = appendListOptionsToSQL(sql, opt)
hosts := []kolide.Host{}
@@ -292,12 +318,10 @@ func (d *Datastore) ListUniqueHostsInLabels(labels []uint) ([]kolide.Host, error
sqlStatement := `
SELECT DISTINCT h.*
FROM label_query_executions lqe
FROM label_membership lm
JOIN hosts h
ON lqe.host_id = h.id
WHERE lqe.label_id IN (?)
AND lqe.matches = 1
AND NOT h.deleted
ON lm.host_id = h.id
WHERE lm.label_id IN (?)
`
query, args, err := sqlx.In(sqlStatement, labels)
if err != nil {
server/datastore/mysql/migrations/data/20170301093653_AddAllHostsToAllHostsLabel.go
+3 -5
View File
@@ -14,14 +14,12 @@ func init() {
func Up_20170301093653(tx *sql.Tx) error {
// Insert any host not currently in 'All Hosts' label into the label
_, err := tx.Exec(`
INSERT IGNORE INTO label_query_executions (
INSERT IGNORE INTO label_membership (
host_id,
label_id,
matches
label_id
) SELECT
id as host_id,
(SELECT id as label_id FROM labels WHERE name = 'All Hosts' AND label_type = ?),
true as matches
(SELECT id as label_id FROM labels WHERE name = 'All Hosts' AND label_type = ?)
FROM hosts
`,
kolide.LabelTypeBuiltIn)
server/datastore/mysql/migrations/tables/20200405120000_UpdateLabelStorage.go
+67
View File
@@ -0,0 +1,67 @@
package tables
import (
"database/sql"
"github.com/pkg/errors"
)
func init() {
MigrationClient.AddMigration(Up20200405120000, Down20200405120000)
}
func Up20200405120000(tx *sql.Tx) error {
if _, err := tx.Exec(
"CREATE TABLE `label_membership` (" +
"`created_at` timestamp DEFAULT CURRENT_TIMESTAMP," +
"`updated_at` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP," +
"`label_id` int(10) unsigned NOT NULL," +
"`host_id` int(10) unsigned NOT NULL," +
"PRIMARY KEY (`host_id`, `label_id`)," +
"INDEX `idx_lm_label_id` (`label_id`)," +
"CONSTRAINT `fk_lm_host_id` FOREIGN KEY (`host_id`) REFERENCES `hosts` (`id`) ON DELETE CASCADE ON UPDATE CASCADE," +
"CONSTRAINT `fk_lm_label_id` FOREIGN KEY (`label_id`) REFERENCES `labels` (`id`) ON DELETE CASCADE ON UPDATE CASCADE" +
") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;",
); err != nil {
return errors.Wrap(err, "create label_membership table")
}
if _, err := tx.Exec(
"INSERT IGNORE INTO `label_membership` " +
"(`created_at`, `updated_at`, `label_id`, `host_id`) " +
"SELECT `created_at`, `updated_at`, `label_id`, `host_id` " +
"FROM `label_query_executions` WHERE matches",
); err != nil {
return errors.Wrap(err, "copy data from label_query_executions")
}
if _, err := tx.Exec(
"INSERT IGNORE INTO `label_membership` " +
"(`host_id`, `label_id`) " +
"SELECT `id`, (SELECT `id` FROM labels WHERE `name` = 'All Hosts' AND `label_type` = 1) FROM `hosts`",
); err != nil {
return errors.Wrap(err, "ensure all hosts are in all hosts label")
}
if _, err := tx.Exec(
"DROP TABLE `label_query_executions`",
); err != nil {
return errors.Wrap(err, "drop label_query_executions")
}
// MySQL is really particular about using zero values or old values for
// timestamps, so we set a default value that is plenty far in the past, but
// hopefully accepted by most MySQL configurations.
if _, err := tx.Exec(
"ALTER TABLE `hosts` " +
"ADD COLUMN `label_update_time` timestamp NOT NULL DEFAULT '2000-01-01 00:00:00'",
); err != nil {
return errors.Wrap(err, "drop label_query_executions")
}
return nil
}
func Down20200405120000(tx *sql.Tx) error {
return nil
}
server/datastore/mysql/packs.go
+6 -8
View File
@@ -407,14 +407,13 @@ func (d *Datastore) ListPacksForHost(hid uint) ([]*kolide.Pack, error) {
FROM
((SELECT p.* FROM packs p
JOIN pack_targets pt
JOIN label_query_executions lqe
JOIN label_membership lm
ON (
p.id = pt.pack_id
AND pt.target_id = lqe.label_id
AND pt.target_id = lm.label_id
AND pt.type = ?
AND lqe.matches
)
WHERE lqe.host_id = ? AND NOT p.disabled)
WHERE lm.host_id = ? AND NOT p.disabled)
UNION ALL
(SELECT p.*
FROM packs p
@@ -435,11 +434,10 @@ func (d *Datastore) ListHostsInPack(pid uint, opt kolide.ListOptions) ([]uint, e
SELECT DISTINCT h.id
FROM hosts h
JOIN pack_targets pt
JOIN label_query_executions lqe
JOIN label_membership lm
ON (
pt.target_id = lqe.label_id
AND lqe.host_id = h.id
AND lqe.matches
pt.target_id = lm.label_id
AND lm.host_id = h.id
AND pt.type = ?
) OR (
pt.target_id = h.id
server/datastore/mysql/targets.go
+2 -2
View File
@@ -26,7 +26,7 @@ func (d *Datastore) CountHostsInTargets(hostIDs []uint, labelIDs []uint, now tim
COALESCE(SUM(CASE WHEN DATE_ADD(seen_time, INTERVAL LEAST(distributed_interval, config_tls_refresh) + %d SECOND) > ? THEN 1 ELSE 0 END), 0) online,
COALESCE(SUM(CASE WHEN DATE_ADD(created_at, INTERVAL 1 DAY) >= ? THEN 1 ELSE 0 END), 0) new
FROM hosts h
WHERE (id IN (?) OR (id IN (SELECT DISTINCT host_id FROM label_query_executions WHERE label_id IN (?) AND matches = 1)))
WHERE (id IN (?) OR (id IN (SELECT DISTINCT host_id FROM label_membership WHERE label_id IN (?))))
AND NOT deleted
`, kolide.OnlineIntervalBuffer, kolide.OnlineIntervalBuffer)
@@ -66,7 +66,7 @@ func (d *Datastore) HostIDsInTargets(hostIDs []uint, labelIDs []uint) ([]uint, e
sql := `
SELECT DISTINCT id
FROM hosts
WHERE (id IN (?) OR (id IN (SELECT host_id FROM label_query_executions WHERE label_id IN (?) AND matches = 1)))
WHERE (id IN (?) OR (id IN (SELECT host_id FROM label_membership WHERE label_id IN (?))))
ORDER BY id ASC
`
server/kolide/campaigns.go
+7 -10
View File
@@ -25,16 +25,13 @@ type CampaignStore interface {
// distributed query campaign
NewDistributedQueryCampaignTarget(target *DistributedQueryCampaignTarget) (*DistributedQueryCampaignTarget, error)
// CleanupDistributedQueryCampaigns will clean and trim metadata for
// old distributed query campaigns. Any campaign in the QueryWaiting
// state will be moved to QueryComplete after one minute. Any campaign
// in the QueryRunning state will be moved to QueryComplete after one
// day. Any campaign in the QueryComplete state will have the
// associated executions deleted. All times are from creation time. The
// now parameter makes this method easier to test. The return values
// indicate how many campaigns were expired, how many executions were
// deleted, and any error.
CleanupDistributedQueryCampaigns(now time.Time) (expired uint, deleted uint, err error)
// CleanupDistributedQueryCampaigns will clean and trim metadata for old
// distributed query campaigns. Any campaign in the QueryWaiting state will
// be moved to QueryComplete after one minute. Any campaign in the
// QueryRunning state will be moved to QueryComplete after one day. Times
// are from creation time. The now parameter makes this method easier to
// test. The return values indicate how many campaigns were expired and any error.
CleanupDistributedQueryCampaigns(now time.Time) (expired uint, err error)
}
// CampaignService defines the distributed query campaign related service
server/kolide/hosts.go
+1 -4
View File
@@ -65,10 +65,6 @@ type HostStore interface {
// GenerateHostStatusStatistics retrieves the count of online, offline,
// MIA and new hosts.
GenerateHostStatusStatistics(now time.Time) (online, offline, mia, new uint, err error)
// DistributedQueriesForHost retrieves the distributed queries that the
// given host should run. The result map is a mapping from campaign ID
// to query text.
DistributedQueriesForHost(host *Host) (map[uint]string, error)
// HostIDsByName Retrieve the IDs associated with the given hostnames
HostIDsByName(hostnames []string) ([]uint, error)
}
@@ -95,6 +91,7 @@ type Host struct {
// a GUID or a Host Name, but in either case, it MUST be unique
OsqueryHostID string `json:"-" db:"osquery_host_id"`
DetailUpdateTime time.Time `json:"detail_updated_at" db:"detail_update_time"` // Time that the host details were last updated
LabelUpdateTime time.Time `json:"label_updated_at" db:"label_update_time"` // Time that the host details were last updated
SeenTime time.Time `json:"seen_time" db:"seen_time"` // Time that the host was last "seen"
NodeKey string `json:"-" db:"node_key"`
HostName string `json:"hostname" db:"host_name"` // there is a fulltext index on this field
server/mock/datastore_campaigns.go
+2 -2
View File
@@ -20,7 +20,7 @@ type DistributedQueryCampaignTargetIDsFunc func(id uint) (hostIDs []uint, labelI
type NewDistributedQueryCampaignTargetFunc func(target *kolide.DistributedQueryCampaignTarget) (*kolide.DistributedQueryCampaignTarget, error)
type CleanupDistributedQueryCampaignsFunc func(now time.Time) (expired uint, deleted uint, err error)
type CleanupDistributedQueryCampaignsFunc func(now time.Time) (expired uint, err error)
type CampaignStore struct {
NewDistributedQueryCampaignFunc NewDistributedQueryCampaignFunc
@@ -67,7 +67,7 @@ func (s *CampaignStore) NewDistributedQueryCampaignTarget(target *kolide.Distrib
return s.NewDistributedQueryCampaignTargetFunc(target)
}
func (s *CampaignStore) CleanupDistributedQueryCampaigns(now time.Time) (expired uint, deleted uint, err error) {
func (s *CampaignStore) CleanupDistributedQueryCampaigns(now time.Time) (expired uint, err error) {
s.CleanupDistributedQueryCampaignsFuncInvoked = true
return s.CleanupDistributedQueryCampaignsFunc(now)
}
server/service/service_osquery.go
+1
View File
@@ -669,6 +669,7 @@ func (svc service) SubmitDistributedQueryResults(ctx context.Context, results ko
}
if len(labelResults) > 0 {
host.LabelUpdateTime = svc.clock.Now()
err = svc.ds.RecordLabelQueryExecutions(&host, labelResults, svc.clock.Now())
if err != nil {
return osqueryError{message: "failed to save labels: " + err.Error()}
server/service/service_osquery_test.go
+2
View File
@@ -349,6 +349,7 @@ func TestLabelQueries(t *testing.T) {
map[string]kolide.OsqueryStatus{},
)
assert.Nil(t, err)
host.LabelUpdateTime = mockClock.Now()
assert.Equal(t, host, gotHost)
assert.Equal(t, mockClock.Now(), gotTime)
if assert.Len(t, gotResults, 1) {
@@ -367,6 +368,7 @@ func TestLabelQueries(t *testing.T) {
map[string]kolide.OsqueryStatus{},
)
assert.Nil(t, err)
host.LabelUpdateTime = mockClock.Now()
assert.Equal(t, host, gotHost)
assert.Equal(t, mockClock.Now(), gotTime)
if assert.Len(t, gotResults, 2) {
server/test/new_objects.go
+12
View File
@@ -75,6 +75,17 @@ func AddLabelToCampaign(t *testing.T, ds kolide.Datastore, campaignID, labelID u
require.Nil(t, err)
}
func AddAllHostsLabel(t *testing.T, ds kolide.Datastore) {
_, err := ds.NewLabel(
&kolide.Label{
Name: "All Hosts",
Query: "select 1",
LabelType: kolide.LabelTypeBuiltIn,
},
)
require.Nil(t, err)
}
func NewHost(t *testing.T, ds kolide.Datastore, name, ip, key, uuid string, now time.Time) *kolide.Host {
osqueryHostID, _ := kolide.RandomText(10)
h, err := ds.NewHost(&kolide.Host{
@@ -82,6 +93,7 @@ func NewHost(t *testing.T, ds kolide.Datastore, name, ip, key, uuid string, now
NodeKey: key,
UUID: uuid,
DetailUpdateTime: now,
LabelUpdateTime: now,
SeenTime: now,
OsqueryHostID: osqueryHostID,
})