Statestore prototype

NOTICE.txt

@@ -959,6 +959,15 @@ 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.
 
+--------------------------------------------------------------------
+Dependency: github.com/elastic/go-concert
+Revision: fde4d7dd2a8c9d3f3c1fc9d22c09dace470433ae
+License type (autodetected): Apache-2.0
+./vendor/github.com/elastic/go-concert/LICENSE:
+--------------------------------------------------------------------
+Apache License 2.0
+
+
 --------------------------------------------------------------------
 Dependency: github.com/elastic/go-libaudit
 Version: v0.4.0

filebeat/input/v2/statestore/connector.go

@@ -0,0 +1,98 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 statestore
+
+import (
+	"sync"
+
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/registry"
+)
+
+// Connector is used to connect to a store backed by a registry.
+type Connector struct {
+	log      *logp.Logger
+	registry *registry.Registry
+
+	mux sync.Mutex
+
+	// set of stores currently with at least one active session.
+	stores map[string]*sharedStore
+}
+
+// NewConnector creates a new store connector for accessing a resource Store.
+func NewConnector(log *logp.Logger, reg *registry.Registry) *Connector {
+	invariant(log != nil, "missing logger")
+	invariant(reg != nil, "missing registry")
+
+	return &Connector{
+		log:      log,
+		registry: reg,
+		stores:   map[string]*sharedStore{},
+	}
+}
+
+// Open creates a connection to a named store.
+func (c *Connector) Open(name string) (*Store, error) {
+	ok := false
+
+	c.mux.Lock()
+	defer c.mux.Unlock()
+
+	persistentStore, err := c.registry.Get(name)
+	if err != nil {
+		return nil, err
+	}
+	defer ifNotOK(&ok, func() {
+		persistentStore.Close()
+	})
+
+	shared := c.stores[name]
+	if shared == nil {
+		shared = &sharedStore{
+			name:            name,
+			persistentStore: persistentStore,
+			resources:       table{},
+		}
+		c.stores[name] = shared
+	} else {
+		shared.refCount.Retain()
+	}
+
+	ok = true
+	return newStore(newSession(c, shared)), nil
+}
+
+func (c *Connector) releaseStore(store *sharedStore) {
+	c.mux.Lock()
+	released := store.refCount.Release()
+	if released {
+		delete(c.stores, store.name)
+	}
+	c.mux.Unlock()
+
+	if released {
+		store.close()
+	}
+}
+
+func ifNotOK(b *bool, fn func()) {
+	if !(*b) {
+		fn()
+	}
+}

filebeat/input/v2/statestore/errors.go

@@ -0,0 +1,93 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 statestore
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
+
+// error codes
+
+var (
+	// ErrClosed error code indicates that the operation can not
+	// be executed, because the store has been closed already.
+	ErrClosed = errors.New("store is closed")
+)
+
+// Error provides a common error type used by the statestore package.  It
+// reports the failed operation, custom message and the root cause if
+// available.
+type Error struct {
+	op      string
+	code    error
+	message string
+	cause   error
+}
+
+// Op returns the name of the operation that failed.
+func (e *Error) Op() string {
+	return e.op
+}
+
+// Code returns a sentinal error that can be used for checking the error type.
+func (e *Error) Code() error {
+	return e.code
+}
+
+// Unwrap returns the cause if available.
+func (e *Error) Unwrap() error {
+	return e.cause
+}
+
+// Error builds the complete error string.
+func (e *Error) Error() string {
+	var buf strings.Builder
+
+	pad := func() {
+		if buf.Len() > 0 {
+			buf.WriteString(": ")
+		}
+	}
+
+	padOpt := func(err error) {
+		if err != nil {
+			pad()
+			fmt.Fprintf(&buf, "%+v", err)
+		}
+	}
+
+	if e.op != "" {
+		buf.WriteString(e.op)
+	}
+	padOpt(e.code)
+	if e.message != "" {
+		pad()
+		buf.WriteString(e.message)
+	}
+	padOpt(e.cause)
+	return buf.String()
+}
+
+func raiseClosed(op string) *Error {
+	return &Error{
+		op:   op,
+		code: ErrClosed,
+	}
+}

filebeat/input/v2/statestore/lockmode.go

@@ -0,0 +1,46 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 statestore
+
+import (
+	"fmt"
+	"sync"
+)
+
+type lockMode int
+
+const (
+	lockRequired lockMode = iota
+	lockAlreadyTaken
+	lockMustRelease
+)
+
+func withLockMode(mux *sync.Mutex, lm lockMode, fn func() error) error {
+	switch lm {
+	case lockRequired:
+		mux.Lock()
+		defer mux.Unlock()
+	case lockMustRelease:
+		defer mux.Unlock()
+	case lockAlreadyTaken:
+	default:
+		panic(fmt.Errorf("unknown lock mode: %v", lm))
+	}
+
+	return fn()
+}

filebeat/input/v2/statestore/op.go

@@ -0,0 +1,100 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 statestore
+
+import (
+	"runtime"
+
+	"github.com/elastic/beats/libbeat/registry"
+)
+
+// ResourceUpdateOp defers a state update to be written to the persistent store.
+// The operation can be applied to the registry using ApplyWith. Calling Close
+// will mark the operation as done.
+type ResourceUpdateOp struct {
+	session *storeSession
+	key     ResourceKey
+	entry   *resourceEntry
+	updates interface{}
+	applied bool
+}
+
+func newUpdateOp(session *storeSession, key ResourceKey, entry *resourceEntry, updates interface{}) *ResourceUpdateOp {
+	session.Retain()
+	entry.refCount.Retain()
+
+	op := &ResourceUpdateOp{
+		session: session,
+		key:     key,
+		entry:   entry,
+		updates: updates,
+	}
+
+	runtime.SetFinalizer(op, (*ResourceUpdateOp).finalize)
+	return op
+}
+
+// ApplyWith applies the operation using the withTx helper function. The helper
+// function is responsible for creating and maintaining a write transaction for
+// the provided store.  If possible the helper should keep the transaction open
+// if an array of operations are applied.
+func (op *ResourceUpdateOp) ApplyWith(withTx func(*registry.Store, func(*registry.Tx) error) error) error {
+	store := op.session.store
+	err := withTx(store.persistentStore, func(tx *registry.Tx) error {
+		return tx.Update(registry.Key(op.key), op.updates)
+	})
+	op.applied = true
+	return err
+}
+
+// Close marks the operation as done. ApplyWith can not be run anymore
+// afterwards.  If all pending operations have been closed, the persistent
+// store is assumed to be in sync with the in memory state.
+func (op *ResourceUpdateOp) Close() {
+	runtime.SetFinalizer(op, nil)
+	op.finalize()
+}
+
+func (op *ResourceUpdateOp) finalize() {
+	if !op.applied {
+		panic("unapplied resource update detected")
+	}
+
+	op.closePending()
+	op.unlink()
+	op.session.Release()
+}
+
+func (op *ResourceUpdateOp) closePending() {
+	entry := op.entry
+
+	entry.value.mux.Lock()
+	defer entry.value.mux.Unlock()
+
+	invariant(entry.value.pending > 0, "there should be pending updates")
+	entry.value.pending--
+	if entry.value.pending == 0 {
+		// we are in sync now, let's remove duplicate data from main memory.
+		entry.value.cached = nil
+	}
+}
+
+func (op *ResourceUpdateOp) unlink() {
+	store, entry := op.session.store, op.entry
+	store.releaseEntry(entry)
+}