Merge pull request #204 from prometheus/fix-wal-closing

Close WAL when closing the DB

db.go

@@ -101,7 +101,6 @@ type DB struct {
 	opts      *Options
 	chunkPool chunks.Pool
 	compactor Compactor
-	wal       WAL
 
 	// Mutex for that must be held when modifying the general block layout.
 	mtx    sync.RWMutex
@@ -572,6 +571,7 @@ func (db *DB) Close() error {
 	if db.lockf != nil {
 		merr.Add(db.lockf.Unlock())
 	}
+	merr.Add(db.head.Close())
 	return merr.Err()
 }
 

db_test.go

@@ -524,3 +524,30 @@ func TestDB_e2e(t *testing.T) {
 
 	return
 }
+
+func TestWALFlushedOnDBClose(t *testing.T) {
+	tmpdir, _ := ioutil.TempDir("", "test")
+	defer os.RemoveAll(tmpdir)
+
+	db, err := Open(tmpdir, nil, nil, nil)
+	require.NoError(t, err)
+
+	lbls := labels.Labels{labels.Label{Name: "labelname", Value: "labelvalue"}}
+
+	app := db.Appender()
+	_, err = app.Add(lbls, 0, 1)
+	require.NoError(t, err)
+	require.NoError(t, app.Commit())
+
+	db.Close()
+
+	db, err = Open(tmpdir, nil, nil, nil)
+	require.NoError(t, err)
+
+	q, err := db.Querier(0, 1)
+	require.NoError(t, err)
+
+	values, err := q.LabelValues("labelname")
+	require.NoError(t, err)
+	require.Equal(t, values, []string{"labelvalue"})
+}

head.go

@@ -739,6 +739,11 @@ func (h *Head) MaxTime() int64 {
 	return atomic.LoadInt64(&h.maxTime)
 }
 
+// Close flushes the WAL and closes the head.
+func (h *Head) Close() error {
+	return h.wal.Close()
+}
+
 type headChunkReader struct {
 	head       *Head
 	mint, maxt int64

head_test.go

@@ -37,6 +37,7 @@ func BenchmarkCreateSeries(b *testing.B) {
 	if err != nil {
 		require.NoError(b, err)
 	}
+	defer h.Close()
 
 	b.ReportAllocs()
 	b.ResetTimer()
@@ -131,6 +132,7 @@ func TestHead_ReadWAL(t *testing.T) {
 
 	head, err := NewHead(nil, nil, wal, 1000)
 	require.NoError(t, err)
+	defer head.Close()
 
 	require.NoError(t, head.ReadWAL())
 	require.Equal(t, uint64(100), head.lastSeriesID)
@@ -163,6 +165,7 @@ func TestHead_ReadWAL(t *testing.T) {
 func TestHead_Truncate(t *testing.T) {
 	h, err := NewHead(nil, nil, nil, 1000)
 	require.NoError(t, err)
+	defer h.Close()
 
 	h.initTime(0)
 
@@ -275,6 +278,7 @@ func TestHeadDeleteSimple(t *testing.T) {
 
 	head, err := NewHead(nil, nil, nil, 1000)
 	require.NoError(t, err)
+	defer head.Close()
 
 	app := head.Appender()