add-go-fuse-to-inject-filesystem-error

.travis.yml

@@ -1,20 +1,37 @@
 dist: trusty
 language: go
-os:
-  - windows
-  - linux
-  - osx
-
-go:
-  - 1.12.x
-
 go_import_path: github.com/prometheus/tsdb
+matrix:
+  include:
+  - os: linux
+    dist: trusty
+    sudo: required
+    go: 1.12.x
+    install:
+      - make deps
+    script:
+      - sudo -E apt-get -yq --no-install-suggests --no-install-recommends --force-yes install fuse
+      - sudo modprobe fuse
+      - sudo chmod a+rw /etc/fuse.conf
+      - sudo echo user_allow_other >> /etc/fuse.conf
+      - make all
+  - os: osx
+    go: 1.12.x
+    sudo: required
+    osx_image: xcode8.3
+    script:
+      - brew update
+      - brew tap homebrew/cask
+      - brew cask install osxfuse
+      - sudo /Library/Filesystems/osxfuse.fs/Contents/Resources/load_osxfuse
+      - sudo sysctl -w vfs.generic.osxfuse.tunables.allow_other=1
+      - make all
+  - os: windows
+    go: 1.12.x
+    before_install:
+      - choco install make
+    install:
+      - make deps
+    script:
+      - make test
 
-before_install:
-  - if [[ "$TRAVIS_OS_NAME" == "windows" ]]; then choco install make; fi
-
-install:
-  - make deps
-
-script:
-  - if [[ "$TRAVIS_OS_NAME" == "windows" ]]; then make test; else make all; fi

block_test.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
+	"github.com/prometheus/tsdb/testutil/fuse"
 	"io/ioutil"
 	"math/rand"
 	"os"
@@ -292,3 +293,40 @@ func populateSeries(lbls []map[string]string, mint, maxt int64) []Series {
 	}
 	return series
 }
+
+// TestFailedDelete ensures that the block is in its original state when a delete fails.
+func TestFailedDelete(t *testing.T) {
+	original, err := ioutil.TempDir("", "original")
+	testutil.Ok(t, err)
+	mountpoint, err := ioutil.TempDir("", "mountpoint")
+	testutil.Ok(t, err)
+
+	defer func() {
+		testutil.Ok(t, os.RemoveAll(mountpoint))
+		testutil.Ok(t, os.RemoveAll(original))
+	}()
+
+	_, file := filepath.Split(createBlock(t, original, genSeries(1, 1, 1, 100)))
+	server, err := fuse.NewServer(original, mountpoint, fuse.FailingRenameHook{})
+	if err != nil {
+		t.Skip(err) // Skip the test for any error. These tests are optional.
+	}
+	defer func() {
+		testutil.Ok(t, server.Close())
+	}()
+
+	expHash, err := testutil.DirHash(original)
+	testutil.Ok(t, err)
+	pb, err := OpenBlock(nil, filepath.Join(mountpoint, file), nil)
+	testutil.Ok(t, err)
+	defer func() {
+		testutil.Ok(t, pb.Close())
+	}()
+
+	testutil.NotOk(t, pb.Delete(1, 10, labels.NewMustRegexpMatcher("", ".*")))
+
+	actHash, err := testutil.DirHash(original)
+	testutil.Ok(t, err)
+
+	testutil.Equals(t, expHash, actHash, "the block dir hash has changed after a failed delete")
+}

go.mod

@@ -5,10 +5,11 @@ require (
 	github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954
 	github.com/go-kit/kit v0.8.0
 	github.com/golang/snappy v0.0.1
+	github.com/hanwen/go-fuse v1.0.0
 	github.com/oklog/ulid v1.3.1
 	github.com/pkg/errors v0.8.0
 	github.com/prometheus/client_golang v1.0.0
 	golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4
-	golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5
+	golang.org/x/sys v0.0.0-20190422165155-953cdadca894
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6
 )

go.sum

@@ -27,6 +27,8 @@ github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/hanwen/go-fuse v1.0.0 h1:GxS9Zrn6c35/BnfiVsZVWmsG803xwE7eVRDvcf/BEVc=
+github.com/hanwen/go-fuse v1.0.0/go.mod h1:unqXarDXqzAk0rt98O2tVndEPIpUgLD9+rwFisZH3Ok=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -74,9 +76,12 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f h1:Bl/8QSvNqXvPGPGXa2z5xUTm
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 h1:YUO/7uOKsKeq9UokNS62b8FYywz3ker1l1vDZRCRefw=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5 h1:mzjBh+S5frKOsOBobWIMAbXavqjmgO17k/2puhcFR94=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190422165155-953cdadca894 h1:Cz4ceDQGXuKRnVBDTS23GTn/pU5OE2C0WrNTOYK1Uuc=
+golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

testutil/directory.go

@@ -14,9 +14,12 @@
 package testutil
 
 import (
+	"crypto/md5"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"strconv"
 )
 
 const (
@@ -143,3 +146,38 @@ func DirSize(path string) (int64, error) {
 	})
 	return size, err
 }
+
+// DirHash returns a hash of all files attributes and their content within a directory.
+func DirHash(path string) ([]byte, error) {
+	hash := md5.New()
+	err := filepath.Walk(path, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if !info.IsDir() {
+			f, err := os.Open(path)
+			if err != nil {
+				return err
+			}
+			defer f.Close()
+
+			if _, err := io.Copy(hash, f); err != nil {
+				return err
+			}
+
+			if _, err := io.WriteString(hash, strconv.Itoa(int(info.Size()))); err != nil {
+				return err
+			}
+			if _, err := io.WriteString(hash, info.Name()); err != nil {
+				return err
+			}
+
+			if _, err := io.WriteString(hash, string(info.ModTime().Unix())); err != nil {
+				return err
+			}
+
+		}
+		return err
+	})
+	return hash.Sum(nil), err
+}

testutil/fuse/fuse.go

@@ -0,0 +1,160 @@
+// +build linux darwin
+
+// Copyright 2019 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package fuse
+
+import (
+	"fmt"
+	"path/filepath"
+	"runtime"
+	"syscall"
+	"time"
+
+	"github.com/hanwen/go-fuse/fuse"
+	"github.com/hanwen/go-fuse/fuse/nodefs"
+	"github.com/hanwen/go-fuse/fuse/pathfs"
+)
+
+type loopBackFs struct {
+	pathfs.FileSystem
+}
+
+// hookFs implements the fuse FileSystem interface and allows injecting hooks for different operations.
+// Its main purpose is to provide a way to inject file system errors.
+type hookFs struct {
+	original   string
+	mountpoint string
+	fsName     string
+	loopBackFs
+	hook Hook
+}
+
+// String returns the string representation of the mounted points.
+func (h *hookFs) String() string {
+	return fmt.Sprintf("HookFs{Original=%s, Mountpoint=%s, FsName=%s, Underlying fs=%s, hook=%s}",
+		h.original, h.mountpoint, h.fsName, h.loopBackFs.String(), h.hook)
+}
+
+// Rename calls the injected rename hooks if those exist or the underlying loopback fs Rename api.
+func (h *hookFs) Rename(oldName string, newName string, context *fuse.Context) fuse.Status {
+	preHooked, err := h.hook.PreRename(oldName, newName)
+	if preHooked {
+		if err != nil {
+			return fuse.ToStatus(err)
+		}
+	}
+
+	status := h.loopBackFs.Rename(oldName, newName, context)
+
+	postHooked, err := h.hook.PostRename(oldName, newName)
+	if postHooked {
+		if err != nil {
+			return fuse.ToStatus(err)
+		}
+	}
+	return status
+}
+
+func (h *hookFs) newServer() (*fuse.Server, error) {
+	opts := &nodefs.Options{
+		NegativeTimeout: time.Second,
+		AttrTimeout:     time.Second,
+		EntryTimeout:    time.Second,
+	}
+	pathFsOpts := &pathfs.PathNodeFsOptions{ClientInodes: true}
+	pathFs := pathfs.NewPathNodeFs(h, pathFsOpts)
+	conn := nodefs.NewFileSystemConnector(pathFs.Root(), opts)
+	originalAbs, _ := filepath.Abs(h.original)
+	mOpts := &fuse.MountOptions{
+		AllowOther: true,
+		Name:       h.fsName,
+		FsName:     originalAbs,
+	}
+	return fuse.NewServer(conn.RawFS(), h.mountpoint, mOpts)
+}
+
+// Server is a fuse server proxy.
+type Server struct {
+	server     *fuse.Server
+	original   string
+	mountpoint string
+}
+
+// NewServer creates a fuse server and attaches it to the given `mountpoint` directory.
+// The caller should not forget to close the server to release the mountpoints.
+func NewServer(original, mountpoint string, hook Hook) (*Server, error) {
+	fs := &hookFs{
+		original:   original,
+		mountpoint: mountpoint,
+		fsName:     "hookfs",
+		loopBackFs: loopBackFs{pathfs.NewLoopbackFileSystem(original)},
+		hook:       hook,
+	}
+
+	server, err := fs.newServer()
+	if err != nil {
+		return nil, err
+	}
+
+	// Async start fuse server, and it will be stopped when calling `fuse.Unmount()` method.
+	go func() {
+		server.Serve()
+	}()
+
+	return &Server{
+		server:     server,
+		mountpoint: mountpoint,
+		original:   original,
+	}, server.WaitMount()
+}
+
+// Close releases the mountpoints and return false if the unmount fails.
+// When the mountpoint has open files it tries to force unmount.
+func (s *Server) Close() error {
+	err := s.server.Unmount()
+	if err != nil {
+		return s.forceUnmount()
+	}
+
+	return nil
+}
+
+// forceUnmount forces to unmount even when there are still open files.
+func (s *Server) forceUnmount() (err error) {
+	delay := time.Duration(0)
+	for try := 0; try < 5; try++ {
+		err = syscall.Unmount(s.mountpoint, unmountFlag())
+		if err == nil {
+			break
+		}
+
+		// Sleep for a bit. This is not pretty, but there is
+		// no way we can be certain that the kernel thinks all
+		// open files have already been closed.
+		delay = 2*delay + 10*time.Millisecond
+		time.Sleep(delay)
+	}
+
+	return err
+}
+
+// unmountFlag returns platform dependent force unmount flag.
+func unmountFlag() int {
+	if runtime.GOOS == "darwin" {
+		return -1
+	}
+
+	return 0x1
+}

testutil/fuse/fuse_windows.go

@@ -0,0 +1,30 @@
+// Copyright 2019 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package fuse
+
+import "errors"
+
+// Server implements a mock server to allow skipping tests for windows platform
+// which doesn't support fuse.
+type Server struct {
+}
+
+// NewServer always returns an error because fuse does not support Windows platform.
+func NewServer(original, mountpoint string, hook Hook) (*Server, error) {
+	return nil, errors.New("fuse not supported under Windows")
+}
+
+// Close is a noop method.
+func (s *Server) Close() error {
+	return nil
+}