Repository: kolide/launcher

Issue:
Recover corrupt kv.sqlite
I see a decent number of errors when the watchdog iterates sqlite logs: `issuing foreach query: database disk image is malformed (11)`.

We should be able to detect and attempt to recover from this state. If we're unable to recover, we should re-create the database instead.

Hints from issue comments:

Solution patch:
diff --git a/ee/agent/types/mocks/logstore.go b/ee/agent/types/mocks/logstore.go
new file mode 100644
index 00000000..04c0d936
--- /dev/null
+++ b/ee/agent/types/mocks/logstore.go
@@ -0,0 +1,98 @@
+// Code generated by mockery v2.53.5. DO NOT EDIT.
+
+package mocks
+
+import mock "github.com/stretchr/testify/mock"
+
+// TimestampedIteratorDeleterAppenderCloser is an autogenerated mock type for the TimestampedIteratorDeleterAppenderCloser type
+type TimestampedIteratorDeleterAppenderCloser struct {
+	mock.Mock
+}
+
+// AppendValue provides a mock function with given fields: timestamp, value
+func (_m *TimestampedIteratorDeleterAppenderCloser) AppendValue(timestamp int64, value []byte) error {
+	ret := _m.Called(timestamp, value)
+
+	if len(ret) == 0 {
+		panic("no return value specified for AppendValue")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(int64, []byte) error); ok {
+		r0 = rf(timestamp, value)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Close provides a mock function with no fields
+func (_m *TimestampedIteratorDeleterAppenderCloser) Close() error {
+	ret := _m.Called()
+
+	if len(ret) == 0 {
+		panic("no return value specified for Close")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// DeleteRows provides a mock function with given fields: rowids
+func (_m *TimestampedIteratorDeleterAppenderCloser) DeleteRows(rowids ...interface{}) error {
+	var _ca []interface{}
+	_ca = append(_ca, rowids...)
+	ret := _m.Called(_ca...)
+
+	if len(ret) == 0 {
+		panic("no return value specified for DeleteRows")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(...interface{}) error); ok {
+		r0 = rf(rowids...)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// ForEach provides a mock function with given fields: fn
+func (_m *TimestampedIteratorDeleterAppenderCloser) ForEach(fn func(int64, int64, []byte) error) error {
+	ret := _m.Called(fn)
+
+	if len(ret) == 0 {
+		panic("no return value specified for ForEach")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(func(int64, int64, []byte) error) error); ok {
+		r0 = rf(fn)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// NewTimestampedIteratorDeleterAppenderCloser creates a new instance of TimestampedIteratorDeleterAppenderCloser. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+// The first argument is typically a *testing.T value.
+func NewTimestampedIteratorDeleterAppenderCloser(t interface {
+	mock.TestingT
+	Cleanup(func())
+}) *TimestampedIteratorDeleterAppenderCloser {
+	mock := &TimestampedIteratorDeleterAppenderCloser{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
diff --git a/ee/watchdog/controller_windows.go b/ee/watchdog/controller_windows.go
index 238cd133..0754dcef 100644
--- a/ee/watchdog/controller_windows.go
+++ b/ee/watchdog/controller_windows.go
@@ -208,6 +208,20 @@ func (wc *WatchdogController) publishLogs(ctx context.Context) {
 		return nil
 	}); err != nil {
 		wc.slogger.Log(ctx, slog.LevelError, "iterating sqlite logs", "err", err)
+
+		// Close and try to reopen - validatedDbConn will handle full corruption
+		if err := wc.logPublisher.Close(); err != nil {
+			wc.slogger.Log(ctx, slog.LevelError, "error closing log publisher after error", "err", err)
+		}
+
+		newPublisher, err := agentsqlite.OpenRW(ctx, wc.knapsack.RootDirectory(), agentsqlite.WatchdogLogStore)
+		if err != nil {
+			wc.slogger.Log(ctx, slog.LevelError, "error opening log publisher after error", "err", err)
+		} else {
+			wc.logPublisher = newPublisher
+		}
+
+		// processing will pick back up on the next iteration
 		return
 	}
 



Test patch:
diff --git a/ee/watchdog/controller_windows_test.go b/ee/watchdog/controller_windows_test.go
index 800c2ab1..578ae500 100644
--- a/ee/watchdog/controller_windows_test.go
+++ b/ee/watchdog/controller_windows_test.go
@@ -13,7 +13,9 @@ import (
 
 	"github.com/kolide/launcher/ee/agent/flags/keys"
 	typesmocks "github.com/kolide/launcher/ee/agent/types/mocks"
+	"github.com/kolide/launcher/pkg/log/multislogger"
 	"github.com/kolide/launcher/pkg/threadsafebuffer"
+	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 )
 
@@ -57,10 +59,12 @@ func TestInterrupt_Multiple(t *testing.T) {
 	mockKnapsack.On("Slogger").Return(testSlogger)
 	mockKnapsack.On("Identifier").Return("kolide-k2").Maybe()
 	mockKnapsack.On("KolideServerURL").Return("k2device.kolide.com")
-	mockKnapsack.On("LauncherWatchdogDisabled").Return(true).Maybe()
+	mockKnapsack.On("LauncherWatchdogDisabled").Return(false).Maybe()
 
 	controller, _ := NewController(t.Context(), mockKnapsack, "")
 
+	require.True(t, controller.shouldManageWatchdog(), "could not manage watchdog, running in admin mode?")
+
 	// Let the handler run for a bit
 	go controller.Run()
 	time.Sleep(3 * time.Second)
@@ -159,6 +163,8 @@ func TestFlagsChanged(t *testing.T) {
 				taskManager: &mockTaskManager{slogger: testSlogger},
 			}
 
+			require.True(t, controller.shouldManageWatchdog(), "could not manage watchdog, running in admin mode?")
+
 			controller.taskManager.setCurrentInstallState(tt.initialInstallState)
 			controller.FlagsChanged(t.Context(), keys.LauncherWatchdogDisabled)
 			require.Equal(t, tt.finalInstallState, controller.taskManager.currentInstallState())
@@ -176,3 +182,79 @@ func TestFlagsChanged(t *testing.T) {
 		})
 	}
 }
+
+func TestPublishLogs(t *testing.T) {
+	t.Parallel()
+	testSlogger := multislogger.NewNopLogger()
+	tempRootDir := t.TempDir()
+
+	mockKnapsack := typesmocks.NewKnapsack(t)
+	mockKnapsack.On("KolideServerURL").Return("k2device.kolide.com")
+	mockKnapsack.On("LauncherWatchdogDisabled").Return(false)
+	mockKnapsack.On("RootDirectory").Return(tempRootDir)
+
+	t.Run("successfully publishes and deletes logs", func(t *testing.T) {
+		t.Parallel()
+
+		mockPublisher := typesmocks.NewTimestampedIteratorDeleterAppenderCloser(t)
+		mockPublisher.On("ForEach", mock.Anything).Run(func(args mock.Arguments) {
+			fn := args.Get(0).(func(int64, int64, []byte) error)
+			fn(1, 1000, []byte(`{"msg":"test log 1"}`))
+			fn(2, 2000, []byte(`{"msg":"test log 2"}`))
+		}).Return(nil)
+		mockPublisher.On("DeleteRows", int64(1), int64(2)).Return(nil)
+
+		controller := &WatchdogController{
+			slogger:      testSlogger,
+			knapsack:     mockKnapsack,
+			interrupt:    make(chan struct{}, 1),
+			logPublisher: mockPublisher,
+			taskManager:  &mockTaskManager{slogger: testSlogger},
+		}
+
+		controller.publishLogs(t.Context())
+
+		mockPublisher.AssertCalled(t, "DeleteRows", int64(1), int64(2))
+	})
+
+	t.Run("no logs to publish", func(t *testing.T) {
+		t.Parallel()
+
+		mockPublisher := typesmocks.NewTimestampedIteratorDeleterAppenderCloser(t)
+		mockPublisher.On("ForEach", mock.Anything).Return(nil)
+
+		controller := &WatchdogController{
+			slogger:      testSlogger,
+			knapsack:     mockKnapsack,
+			interrupt:    make(chan struct{}, 1),
+			logPublisher: mockPublisher,
+			taskManager:  &mockTaskManager{slogger: testSlogger},
+		}
+
+		controller.publishLogs(t.Context())
+
+		mockPublisher.AssertNotCalled(t, "DeleteRows", "delete rows should not be called for no logs")
+	})
+
+	t.Run("ForEach error", func(t *testing.T) {
+		t.Parallel()
+
+		mockPublisher := typesmocks.NewTimestampedIteratorDeleterAppenderCloser(t)
+		mockPublisher.On("ForEach", mock.Anything).Return(errors.New("some err"))
+		mockPublisher.On("Close").Return(errors.New("some err"))
+
+		controller := &WatchdogController{
+			slogger:      testSlogger,
+			knapsack:     mockKnapsack,
+			interrupt:    make(chan struct{}, 1),
+			logPublisher: mockPublisher,
+			taskManager:  &mockTaskManager{slogger: testSlogger},
+		}
+
+		controller.publishLogs(t.Context())
+
+		mockPublisher.AssertCalled(t, "Close")
+		require.NotEqual(t, mockPublisher, controller.logPublisher, "logPublisher should have been replaced with a new publisher")
+		require.NoError(t, controller.logPublisher.Close())
+	})
+}