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)
+}