Uploaded wrapper source code

Author: odeke-em

README.md

@@ -1,2 +1,2 @@
-# gomongorwapper
+# gomongowrapper
 MongoDB Go wrapper source code

example_test.go

@@ -0,0 +1,52 @@
+// Copyright 2018, OpenCensus 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 mongowrapper_test
+
+import (
+	"context"
+	"log"
+
+	"github.com/mongodb/mongo-go-driver/bson"
+
+	"github.com/opencensus-integrations/gomongowrapper"
+
+	"go.opencensus.io/trace"
+)
+
+func Example() {
+	client, err := mongowrapper.NewClient("mongodb://foo:bar@localhost:27017")
+	if err != nil {
+		log.Fatalf("Failed to create the new client: %v", err)
+	}
+	coll := client.Database("the_db").Collection("music")
+
+	ctx, span := trace.StartSpan(context.Background(), "Fetch")
+	defer span.End()
+
+	q := bson.M{"name": "Examples"}
+	cur, err := coll.Find(ctx, q)
+	if err != nil {
+		log.Fatalf("Find error: %v", err)
+	}
+
+	for cur.Next(ctx) {
+		elem := make(map[string]int)
+		if err := cur.Decode(elem); err != nil {
+			log.Printf("Decode error: %v", err)
+			continue
+		}
+		log.Printf("Got result: %v\n", elem)
+	}
+}

mongo.go

@@ -0,0 +1,40 @@
+// Copyright 2018, OpenCensus 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 mongowrapper
+
+import (
+	"context"
+
+	"github.com/mongodb/mongo-go-driver/mongo"
+	"github.com/mongodb/mongo-go-driver/mongo/clientopt"
+)
+
+func Connect(ctx context.Context, uri string, opts ...clientopt.Option) (*WrappedClient, error) {
+	ctx, span := roundtripTrackingSpan(ctx, "github.com/mongodb/mongo-go-driver.Connect")
+	defer span.end(ctx)
+
+	cc, err := mongo.NewClientWithOptions(uri, opts...)
+	if err != nil {
+		span.setError(err)
+		return nil, err
+	}
+
+	wc := &WrappedClient{cc: cc}
+	err = wc.Connect(ctx)
+	if err != nil {
+		span.setError(err)
+	}
+	return wc, err
+}

observability.go

@@ -0,0 +1,95 @@
+// Copyright 2018, OpenCensus 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 mongowrapper
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+	"go.opencensus.io/trace"
+)
+
+var (
+	keyMethod, _ = tag.NewKey("method")
+	keyStatus, _ = tag.NewKey("status")
+	keyError, _  = tag.NewKey("error")
+)
+
+var mLatencyMs = stats.Float64("latency", "The latency in milliseconds", "ms")
+
+var allViews = []*view.View{
+	{
+		Name: "mongo/client/latency", Description: "The latency of the various calls",
+		Measure: mLatencyMs,
+		// [0ms, 0.001ms, 0.005ms, 0.01ms, 0.05ms, 0.1ms, 0.5ms, 1ms, 1.5ms, 2ms, 2.5ms, 5ms, 10ms, 25ms, 50ms, 100ms, 200ms,
+		//      400ms, 600ms, 800ms, 1s, 1.5s, 2s, 2.5s, 5s, 10s, 20s, 40s, 100s, 200s, 500s, 1000s]
+		//
+		Aggregation: view.Distribution(0, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 1.5, 2, 2.5, 5, 10, 25, 50, 100, 200,
+			400, 600, 800, 1000, 1500, 2000, 2500, 5000, 10000, 20000, 40000, 100000, 200000, 500000, 1000000),
+		TagKeys: []tag.Key{keyMethod, keyStatus, keyError},
+	},
+	{
+		Name: "mongo/client/calls", Description: "The various calls",
+		Measure:     mLatencyMs,
+		Aggregation: view.Count(),
+		TagKeys:     []tag.Key{keyMethod, keyStatus, keyError},
+	},
+}
+
+func RegisterAllViews() error {
+	return view.Register(allViews...)
+}
+
+func UnregisterAllViews() {
+	view.Unregister(allViews...)
+}
+
+type spanWithMetrics struct {
+	startTime time.Time
+	method    string
+	lastErr   error
+	span      *trace.Span
+	endOnce   sync.Once
+}
+
+func roundtripTrackingSpan(ctx context.Context, methodName string) (context.Context, *spanWithMetrics) {
+	ctx, span := trace.StartSpan(ctx, methodName)
+	return ctx, &spanWithMetrics{span: span, startTime: time.Now()}
+}
+
+func (swm *spanWithMetrics) setError(err error) {
+	if err != nil {
+		swm.span.SetStatus(trace.Status{Code: trace.StatusCodeUnknown, Message: err.Error()})
+	}
+	swm.lastErr = err
+}
+
+func (swm *spanWithMetrics) end(ctx context.Context) {
+	swm.endOnce.Do(func() {
+		if err := swm.lastErr; err == nil {
+			ctx, _ = tag.New(ctx, tag.Upsert(keyMethod, swm.method), tag.Upsert(keyStatus, "OK"))
+		} else {
+			ctx, _ = tag.New(ctx, tag.Upsert(keyMethod, swm.method), tag.Upsert(keyError, err.Error()))
+		}
+
+		latencyMs := float64(time.Now().Sub(swm.startTime)) / 1e6
+		stats.Record(ctx, mLatencyMs.M(latencyMs))
+		swm.span.End()
+	})
+}

wrapped_client.go

@@ -0,0 +1,121 @@
+// Copyright 2018, OpenCensus 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 mongowrapper
+
+import (
+	"context"
+
+	"github.com/mongodb/mongo-go-driver/core/readpref"
+	"github.com/mongodb/mongo-go-driver/core/session"
+	"github.com/mongodb/mongo-go-driver/mongo"
+	"github.com/mongodb/mongo-go-driver/mongo/dbopt"
+	"github.com/mongodb/mongo-go-driver/mongo/listdbopt"
+	"github.com/mongodb/mongo-go-driver/mongo/sessionopt"
+)
+
+type WrappedClient struct {
+	cc *mongo.Client
+}
+
+func NewClient(uri string) (*WrappedClient, error) {
+	client, err := mongo.NewClient(uri)
+	if err != nil {
+		return nil, err
+	}
+	return &WrappedClient{cc: client}, nil
+}
+
+func (wc *WrappedClient) Connect(ctx context.Context) error {
+	ctx, span := roundtripTrackingSpan(ctx, "github.com/mongodb/mongo-go-driver.Client.Connect")
+	defer span.end(ctx)
+
+	err := wc.cc.Connect(ctx)
+	if err != nil {
+		span.setError(err)
+	}
+	return err
+}
+
+func (wc *WrappedClient) ConnectionString() string { return wc.cc.ConnectionString() }
+
+func (wc *WrappedClient) Database(name string, opts ...dbopt.Option) *WrappedDatabase {
+	db := wc.cc.Database(name, opts...)
+	if db == nil {
+		return nil
+	}
+	return &WrappedDatabase{db: db}
+}
+
+func (wc *WrappedClient) Disconnect(ctx context.Context) error {
+	ctx, span := roundtripTrackingSpan(ctx, "github.com/mongodb/mongo-go-driver.Client.Disconnect")
+	defer span.end(ctx)
+
+	err := wc.cc.Disconnect(ctx)
+	if err != nil {
+		span.setError(err)
+	}
+	return err
+}
+
+func (wc *WrappedClient) ListDatabaseNames(ctx context.Context, filter interface{}, opts ...listdbopt.ListDatabases) ([]string, error) {
+	ctx, span := roundtripTrackingSpan(ctx, "github.com/mongodb/mongo-go-driver.Client.ListDatabaseNames")
+	defer span.end(ctx)
+
+	dbs, err := wc.cc.ListDatabaseNames(ctx, filter, opts...)
+	if err != nil {
+		span.setError(err)
+	}
+	return dbs, err
+}
+
+func (wc *WrappedClient) ListDatabases(ctx context.Context, filter interface{}, opts ...listdbopt.ListDatabases) (mongo.ListDatabasesResult, error) {
+	ctx, span := roundtripTrackingSpan(ctx, "github.com/mongodb/mongo-go-driver.Client.ListDatabases")
+	defer span.end(ctx)
+
+	dbr, err := wc.cc.ListDatabases(ctx, filter, opts...)
+	if err != nil {
+		span.setError(err)
+	}
+	return dbr, err
+}
+
+func (wc *WrappedClient) Ping(ctx context.Context, rp *readpref.ReadPref) error {
+	ctx, span := roundtripTrackingSpan(ctx, "github.com/mongodb/mongo-go-driver.Client.Ping")
+	defer span.end(ctx)
+
+	err := wc.cc.Ping(ctx, rp)
+	if err != nil {
+		span.setError(err)
+	}
+	return err
+}
+
+func (wc *WrappedClient) StartSession(opts ...sessionopt.Session) (mongo.Session, error) {
+	return wc.cc.StartSession(opts...)
+}
+
+func (wc *WrappedClient) UseSession(ctx context.Context, fn func(mongo.SessionContext) error) error {
+	return wc.cc.UseSession(ctx, fn)
+}
+
+func (wc *WrappedClient) UseSessionWithOptions(ctx context.Context, opts []sessionopt.Session, fn func(mongo.SessionContext) error) error {
+	return wc.cc.UseSessionWithOptions(ctx, opts, fn)
+}
+
+func (wc *WrappedClient) ValidSession(sess *session.Client) error {
+	return wc.cc.ValidSession(sess)
+}
+
+func (wc *WrappedClient) Client() *mongo.Client { return wc.cc }