Fix tsdb code and tests to work on Windows. (#6547)

Add back Windows CI, we lost it when tsdb was merged into the prometheus
repo. There's many tests failing outside tsdb, so only test tsdb for
now.

Fixes #6513

Signed-off-by: Brian Brazil <brian.brazil@robustperception.io>

.circleci/config.yml

@@ -4,6 +4,7 @@ version: 2.1
 orbs:
   prometheus: prometheus/prometheus@0.3.0
   go: circleci/go@0.2.0
+  win: circleci/windows@2.3.0
 
 executors:
   # Whenever the Go version is updated here, .promu.yml
@@ -49,6 +50,13 @@ jobs:
         key: v1-npm-deps-{{ checksum "web/ui/react-app/yarn.lock" }}
         paths:
         - web/ui/react-app/node_modules
+  test_windows:
+    executor: win/default
+    working_directory: /go/src/github.com/prometheus/prometheus
+    steps:
+    - checkout
+    # TSDB is where the most risk is Windows wise, so only test there for now.
+    - run: go test ./tsdb/...
   fuzzit_regression:
     executor: fuzzit
     working_directory: /go/src/github.com/prometheus/prometheus
@@ -78,6 +86,10 @@ workflows:
         filters:
           tags:
             only: /.*/
+    - test_windows:
+        filters:
+          tags:
+            only: /.*/
     - fuzzit_regression:
         filters:
           tags:

tsdb/block_test.go

@@ -197,9 +197,11 @@ func TestCorruptedChunk(t *testing.T) {
 				testutil.Equals(t, test.openErr.Error(), err.Error())
 				return
 			}
+			defer func() { testutil.Ok(t, b.Close()) }()
 
 			querier, err := NewBlockQuerier(b, 0, 1)
 			testutil.Ok(t, err)
+			defer func() { testutil.Ok(t, querier.Close()) }()
 			set, err := querier.Select(labels.MustNewMatcher(labels.MatchEqual, "a", "b"))
 			testutil.Ok(t, err)
 

tsdb/db_test.go

@@ -2655,6 +2655,7 @@ func TestChunkWriter_ReadAfterWrite(t *testing.T) {
 			// Check the content of the chunks.
 			r, err := chunks.NewDirReader(tempDir, nil)
 			testutil.Ok(t, err)
+			defer func() { testutil.Ok(t, r.Close()) }()
 
 			for _, chks := range test.chks {
 				for _, chkExp := range chks {
@@ -2705,4 +2706,5 @@ func TestChunkReader_ConcurrentReads(t *testing.T) {
 		}
 		wg.Wait()
 	}
+	testutil.Ok(t, r.Close())
 }

tsdb/index/index.go

@@ -511,11 +511,11 @@ func (w *Writer) finishSymbols() error {
 		return err
 	}
 
-	var err error
+	sf, err := fileutil.OpenMmapFile(w.f.name)
-	w.symbolFile, err = fileutil.OpenMmapFile(w.f.name)
 	if err != nil {
 		return err
 	}
+	w.symbolFile = sf
 	hash := crc32.Checksum(w.symbolFile.Bytes()[w.toc.Symbols+4:hashPos], castagnoliTable)
 	w.buf1.Reset()
 	w.buf1.PutBE32(hash)
@@ -700,7 +700,11 @@ func (w *Writer) writePostingsOffsetTable() error {
 	if err != nil {
 		return err
 	}
-	defer f.Close()
+	defer func() {
+		if f != nil {
+			f.Close()
+		}
+	}()
 	d := encoding.NewDecbufRaw(realByteSlice(f.Bytes()), int(w.fPO.pos))
 	cnt := w.cntPO
 	for d.Err() == nil && cnt > 0 {
@@ -720,6 +724,10 @@ func (w *Writer) writePostingsOffsetTable() error {
 	}
 
 	// Cleanup temporary file.
+	if err := f.Close(); err != nil {
+		return err
+	}
+	f = nil
 	if err := w.fPO.close(); err != nil {
 		return err
 	}
@@ -962,9 +970,9 @@ type labelIndexHashEntry struct {
 }
 
 func (w *Writer) Close() error {
-	if err := w.ensureStage(idxStageDone); err != nil {
+	// Even if this fails, we need to close all the files.
-		return err
+	ensureErr := w.ensureStage(idxStageDone)
-	}
+
 	if w.symbolFile != nil {
 		if err := w.symbolFile.Close(); err != nil {
 			return err
@@ -980,7 +988,10 @@ func (w *Writer) Close() error {
 			return err
 		}
 	}
-	return w.f.close()
+	if err := w.f.close(); err != nil {
+		return err
+	}
+	return ensureErr
 }
 
 // StringTuples provides access to a sorted list of string tuples.

tsdb/index/index_test.go

@@ -280,6 +280,7 @@ func TestPostingsMany(t *testing.T) {
 
 	ir, err := NewFileReader(fn)
 	testutil.Ok(t, err)
+	defer func() { testutil.Ok(t, ir.Close()) }()
 
 	cases := []struct {
 		in []string

tsdb/wal/watcher_test.go

@@ -427,6 +427,7 @@ func TestReadCheckpointMultipleSegments(t *testing.T) {
 					}
 				}
 			}
+			testutil.Ok(t, w.Close())
 
 			// At this point we should have at least 6 segments, lets create a checkpoint dir of the first 5.
 			checkpointDir := dir + "/wal/checkpoint.000004"