Re-creating mgo sessions in case of errors (read tcp 127.0.0.1:46954->127.0.0.1:27017: i/o timeout)

Question

I wonder about MongoDB session management in Go using mgo, especially about how to correctly ensure a session is closed and how to react on write failures.

I have read the following:

Best practice to maintain a mgo session

Should I copy session for each operation in mgo?

Still, cannot apply it to my situation.

I have two goroutines which store event after event into MongoDB sharing the same *mgo.Session, both looking essiantially like the following:

func storeEvents(session *mgo.Session) {
    session_copy := session.Copy()
    // *** is it correct to defer the session close here? <-----
    defer session_copy.Close()
    col := session_copy.DB("DB_NAME").C("COLLECTION_NAME")
    for {
        event := GetEvent()
        err := col.Insert(&event)
        if err != nil {
            // *** insert FAILED - how to react properly? <-----
            session_copy = session.Copy()
            defer session_copy.Close()
        }
    }
}

col.Insert(&event) after some hours returns the error

read tcp 127.0.0.1:46954->127.0.0.1:27017: i/o timeout

and I am unsure how to react on this. After this error occurs, it occurs on all subsequent writes, hence it seems I have to create a new session. Alternatives for me seem:

1) restart the whole goroutine, i.e.

if err != nil {
    go storeEvents(session)
    return
}

2) create a new session copy

if err != nil {
    session_copy = session.Copy()
    defer session_copy.Close()
    col := session_copy.DB("DB_NAME").C("COLLECTION_NAME")
    continue
}

--> Is it correct how I use defer session_copy.Close()? (Note the above defer references the Close() function of another session. Anyway, those sessions will never be closed since the function never returns. I.e., with time, many sessions will be created and not closed.

Other options?

Answer 1

So I don't know if this is going to help you any, but I don't have any issues with this set up.

I have a mongo package that I import from. This is a template of my mongo.go file

package mongo

import (
    "time"

    "gopkg.in/mgo.v2"
)

var (
    // MyDB ...
    MyDB DataStore
)

// create the session before main starts
func init() {
    MyDB.ConnectToDB()
}

// DataStore containing a pointer to a mgo session
type DataStore struct {
    Session *mgo.Session
}

// ConnectToTagserver is a helper method that connections to pubgears' tagserver
// database
func (ds *DataStore) ConnectToDB() {
    mongoDBDialInfo := &mgo.DialInfo{
        Addrs:    []string{"ip"},
        Timeout:  60 * time.Second,
        Database: "db",
    }
    sess, err := mgo.DialWithInfo(mongoDBDialInfo)
    if err != nil {
        panic(err)
    }
    sess.SetMode(mgo.Monotonic, true)
    MyDB.Session = sess
}

// Close is a helper method that ensures the session is properly terminated
func (ds *DataStore) Close() {
    ds.Session.Close()
}

Then in another package, for example main Updated Based on the comment below

package main

import (
    "../models/mongo"
)

func main() {
    // Grab the main session which was instantiated in the mongo package init function
    sess := mongo.MyDB.Session
    // pass that session in
    storeEvents(sess)
}

func storeEvents(session *mgo.Session) {
    session_copy := session.Copy()
    defer session_copy.Close()

    // Handle panics in a deferred fuction
    // You can turn this into a wrapper (middleware)
    // remove this this function, and just wrap your calls with it, using switch cases
    // you can handle all types of errors
    defer func(session *mgo.Session) {
        if err := recover(); err != nil {
            fmt.Printf("Mongo insert has caused a panic: %s\n", err)
            fmt.Println("Attempting to insert again")
            session_copy := session.Copy()
            defer session_copy.Close()
            col := session_copy.DB("DB_NAME").C("COLLECTION_NAME")
            event := GetEvent()
            err := col.Insert(&event)
            if err != nil {
                fmt.Println("Attempting to insert again failed")
                return
            }
            fmt.Println("Attempting to insert again succesful")
        }
    }(session)

    col := session_copy.DB("DB_NAME").C("COLLECTION_NAME")
    event := GetEvent()
    err := col.Insert(&event)
    if err != nil {
        panic(err)
    }
}

I use a similar setup on my production servers on AWS. I do over 1 million inserts an hour. Hope this helps. Another things I've done to ensure that the mongo servers can handle the connections is increate the ulimit on my production machines. It's talked about in this stack