0187d8f7-4f52-461a-af17-77a5ea7bc7f0

Committed 01 May 2023 08:38PM UTC coverage: 57.253% (+0.04%) from 57.214%

Build # 0187d8f7-4f52-461a-af17-77a5ea7bc7f0

Build Type

push

buildkite

Committed by GitHub

Commit Message

Scaffold config store for SQL (#5239)

Run Details

23 of 23 new or added lines in 3 files covered. (100.0%)

85793 of 149849 relevant lines covered (57.25%)

2412.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.82

/common/persistence/sql/factory.go

// Copyright (c) 2017 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

package sql

import (
        "fmt"
        "sync"

        "github.com/uber/cadence/common/persistence/serialization"

        "github.com/uber/cadence/common/config"
        "github.com/uber/cadence/common/log"
        p "github.com/uber/cadence/common/persistence"
        "github.com/uber/cadence/common/persistence/sql/sqlplugin"
)

type (
        // Factory vends store objects backed by MySQL
        Factory struct {
                cfg         config.SQL
                dbConn      dbConn
                clusterName string
                logger      log.Logger
                parser      serialization.Parser
                dc          *p.DynamicConfiguration
        }

        // dbConn represents a logical mysql connection - its a
        // wrapper around the standard sql connection pool with
        // additional reference counting
        dbConn struct {
                sync.Mutex
                sqlplugin.DB
                refCnt int
                cfg    *config.SQL
        }
)

// NewFactory returns an instance of a factory object which can be used to create
// datastores backed by any kind of SQL store
func NewFactory(
        cfg config.SQL,
        clusterName string,
        logger log.Logger,
        parser serialization.Parser,
        dc *p.DynamicConfiguration,
) *Factory {
        return &Factory{
                cfg:         cfg,
                clusterName: clusterName,
                logger:      logger,
                dbConn:      newRefCountedDBConn(&cfg),
                parser:      parser,
                dc:          dc,
        }
}

// NewTaskStore returns a new task store
func (f *Factory) NewTaskStore() (p.TaskStore, error) {
        conn, err := f.dbConn.get()
        if err != nil {
                return nil, err
        }
        return newTaskPersistence(conn, f.cfg.NumShards, f.logger, f.parser)
}

// NewShardStore returns a new shard store
func (f *Factory) NewShardStore() (p.ShardStore, error) {
        conn, err := f.dbConn.get()
        if err != nil {
                return nil, err
        }
        return NewShardPersistence(conn, f.clusterName, f.logger, f.parser)
}

// NewHistoryStore returns a new history store
func (f *Factory) NewHistoryStore() (p.HistoryStore, error) {
        conn, err := f.dbConn.get()
        if err != nil {
                return nil, err
        }
        return NewHistoryV2Persistence(conn, f.logger, f.parser)
}

// NewDomainStore returns a new metadata store
func (f *Factory) NewDomainStore() (p.DomainStore, error) {
        conn, err := f.dbConn.get()
        if err != nil {
                return nil, err
        }
        return newMetadataPersistenceV2(conn, f.clusterName, f.logger, f.parser)
}

// NewExecutionStore returns an ExecutionStore for a given shardID
func (f *Factory) NewExecutionStore(shardID int) (p.ExecutionStore, error) {
        conn, err := f.dbConn.get()
        if err != nil {
                return nil, err
        }
        return NewSQLExecutionStore(conn, f.logger, shardID, f.parser, f.dc)
}

// NewVisibilityStore returns a visibility store
// TODO sortByCloseTime will be removed and implemented for https://github.com/uber/cadence/issues/3621
func (f *Factory) NewVisibilityStore(sortByCloseTime bool) (p.VisibilityStore, error) {
        return NewSQLVisibilityStore(f.cfg, f.logger)
}

// NewQueue returns a new queue backed by sql
func (f *Factory) NewQueue(queueType p.QueueType) (p.Queue, error) {
        conn, err := f.dbConn.get()
        if err != nil {
                return nil, err
        }

        return newQueueStore(conn, f.logger, queueType)
}

// NewConfigStore returns a new config store backed by sql. Not Yet Implemented.
func (f *Factory) NewConfigStore() (p.ConfigStore, error) {
        conn, err := f.dbConn.get()
        if err != nil {
                return nil, err
        }
        return NewSQLConfigStore(conn, f.logger, f.parser)
}

// Close closes the factory
func (f *Factory) Close() {
        f.dbConn.forceClose()
}

// newRefCountedDBConn returns a  logical mysql connection that
// uses reference counting to decide when to close the
// underlying connection object. The reference count gets incremented
// everytime get() is called and decremented everytime Close() is called
func newRefCountedDBConn(cfg *config.SQL) dbConn {
        return dbConn{cfg: cfg}
}

// get returns a mysql db connection and increments a reference count
// this method will create a new connection, if an existing connection
// does not exist
func (c *dbConn) get() (sqlplugin.DB, error) {
        c.Lock()
        defer c.Unlock()
        if c.refCnt == 0 {
                conn, err := NewSQLDB(c.cfg)
                if err != nil {
                        return nil, err
                }
                c.DB = conn
        }
        c.refCnt++
        return c, nil
}

// forceClose ignores reference counts and shutsdown the underlying connection pool
func (c *dbConn) forceClose() {
        c.Lock()
        defer c.Unlock()
        if c.DB != nil {
                err := c.DB.Close()
                if err != nil {
                        fmt.Println("failed to close database connection, may leak some connection", err)
                }

        }
        c.refCnt = 0
}

// Close closes the underlying connection if the reference count becomes zero
func (c *dbConn) Close() error {
        c.Lock()
        defer c.Unlock()
        c.refCnt--
        if c.refCnt == 0 {
                err := c.DB.Close()
                c.DB = nil
                return err
        }
        return nil
}

1	// Copyright (c) 2017 Uber Technologies, Inc.
2	//
3	// Permission is hereby granted, free of charge, to any person obtaining a copy
4	// of this software and associated documentation files (the "Software"), to deal
5	// in the Software without restriction, including without limitation the rights
6	// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
7	// copies of the Software, and to permit persons to whom the Software is
8	// furnished to do so, subject to the following conditions:
9	//
10	// The above copyright notice and this permission notice shall be included in
11	// all copies or substantial portions of the Software.
12	//
13	// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14	// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15	// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16	// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17	// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
18	// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
19	// THE SOFTWARE.
20
21	package sql
22
23	import (
24	"fmt"
25	"sync"
26
27	"github.com/uber/cadence/common/persistence/serialization"
28
29	"github.com/uber/cadence/common/config"
30	"github.com/uber/cadence/common/log"
31	p "github.com/uber/cadence/common/persistence"
32	"github.com/uber/cadence/common/persistence/sql/sqlplugin"
33	)
34
35	type (
36	// Factory vends store objects backed by MySQL
37	Factory struct {
38	cfg config.SQL
39	dbConn dbConn
40	clusterName string
41	logger log.Logger
42	parser serialization.Parser
43	dc *p.DynamicConfiguration
44	}
45
46	// dbConn represents a logical mysql connection - its a
47	// wrapper around the standard sql connection pool with
48	// additional reference counting
49	dbConn struct {
50	sync.Mutex
51	sqlplugin.DB
52	refCnt int
53	cfg *config.SQL
54	}
55	)
56
57	// NewFactory returns an instance of a factory object which can be used to create
58	// datastores backed by any kind of SQL store
59	func NewFactory(
60	cfg config.SQL,
61	clusterName string,
62	logger log.Logger,
63	parser serialization.Parser,
64	dc *p.DynamicConfiguration,
65	) *Factory {	66✔
66	return &Factory{	66✔
67	cfg: cfg,	66✔
68	clusterName: clusterName,	66✔
69	logger: logger,	66✔
70	dbConn: newRefCountedDBConn(&cfg),	66✔
71	parser: parser,	66✔
72	dc: dc,	66✔
73	}	66✔
74	}	66✔
75
76	// NewTaskStore returns a new task store
77	func (f *Factory) NewTaskStore() (p.TaskStore, error) {	34✔
78	conn, err := f.dbConn.get()	34✔
79	if err != nil {	34✔
80	return nil, err	×
81	}	×
82	return newTaskPersistence(conn, f.cfg.NumShards, f.logger, f.parser)	34✔
83	}
84
85	// NewShardStore returns a new shard store
86	func (f *Factory) NewShardStore() (p.ShardStore, error) {	34✔
87	conn, err := f.dbConn.get()	34✔
88	if err != nil {	34✔
89	return nil, err	×
90	}	×
91	return NewShardPersistence(conn, f.clusterName, f.logger, f.parser)	34✔
92	}
93
94	// NewHistoryStore returns a new history store
95	func (f *Factory) NewHistoryStore() (p.HistoryStore, error) {	34✔
96	conn, err := f.dbConn.get()	34✔
97	if err != nil {	34✔
98	return nil, err	×
99	}	×
100	return NewHistoryV2Persistence(conn, f.logger, f.parser)	34✔
101	}
102
103	// NewDomainStore returns a new metadata store
104	func (f *Factory) NewDomainStore() (p.DomainStore, error) {	34✔
105	conn, err := f.dbConn.get()	34✔
106	if err != nil {	34✔
107	return nil, err	×
108	}	×
109	return newMetadataPersistenceV2(conn, f.clusterName, f.logger, f.parser)	34✔
110	}
111
112	// NewExecutionStore returns an ExecutionStore for a given shardID
113	func (f *Factory) NewExecutionStore(shardID int) (p.ExecutionStore, error) {	42✔
114	conn, err := f.dbConn.get()	42✔
115	if err != nil {	42✔
116	return nil, err	×
117	}	×
118	return NewSQLExecutionStore(conn, f.logger, shardID, f.parser, f.dc)	42✔
119	}
120
121	// NewVisibilityStore returns a visibility store
122	// TODO sortByCloseTime will be removed and implemented for https://github.com/uber/cadence/issues/3621
123	func (f *Factory) NewVisibilityStore(sortByCloseTime bool) (p.VisibilityStore, error) {	18✔
124	return NewSQLVisibilityStore(f.cfg, f.logger)	18✔
125	}	18✔
126
127	// NewQueue returns a new queue backed by sql
128	func (f *Factory) NewQueue(queueType p.QueueType) (p.Queue, error) {	34✔
129	conn, err := f.dbConn.get()	34✔
130	if err != nil {	34✔
131	return nil, err	×
132	}	×
133
134	return newQueueStore(conn, f.logger, queueType)	34✔
135	}
136
137	// NewConfigStore returns a new config store backed by sql. Not Yet Implemented.
138	func (f *Factory) NewConfigStore() (p.ConfigStore, error) {	26✔
139	conn, err := f.dbConn.get()	26✔
140	if err != nil {	26✔
141	return nil, err	×
142	}	×
143	return NewSQLConfigStore(conn, f.logger, f.parser)	26✔
144	}
145
146	// Close closes the factory
147	func (f *Factory) Close() {	34✔
148	f.dbConn.forceClose()	34✔
149	}	34✔
150
151	// newRefCountedDBConn returns a logical mysql connection that
152	// uses reference counting to decide when to close the
153	// underlying connection object. The reference count gets incremented
154	// everytime get() is called and decremented everytime Close() is called
155	func newRefCountedDBConn(cfg *config.SQL) dbConn {	66✔
156	return dbConn{cfg: cfg}	66✔
157	}	66✔
158
159	// get returns a mysql db connection and increments a reference count
160	// this method will create a new connection, if an existing connection
161	// does not exist
162	func (c *dbConn) get() (sqlplugin.DB, error) {	226✔
163	c.Lock()	226✔
164	defer c.Unlock()	226✔
165	if c.refCnt == 0 {	260✔
166	conn, err := NewSQLDB(c.cfg)	34✔
167	if err != nil {	34✔
168	return nil, err	×
169	}	×
170	c.DB = conn	34✔
171	}
172	c.refCnt++	226✔
173	return c, nil	226✔
174	}
175
176	// forceClose ignores reference counts and shutsdown the underlying connection pool
177	func (c *dbConn) forceClose() {	34✔
178	c.Lock()	34✔
179	defer c.Unlock()	34✔
180	if c.DB != nil {	68✔
181	err := c.DB.Close()	34✔
182	if err != nil {	34✔
183	fmt.Println("failed to close database connection, may leak some connection", err)	×
184	}	×
185
186	}
187	c.refCnt = 0	34✔
188	}
189
190	// Close closes the underlying connection if the reference count becomes zero
191	func (c *dbConn) Close() error {	154✔
192	c.Lock()	154✔
193	defer c.Unlock()	154✔
194	c.refCnt--	154✔
195	if c.refCnt == 0 {	154✔
196	err := c.DB.Close()	×
197	c.DB = nil	×
198	return err	×
199	}	×
200	return nil	154✔
201	}

uber / cadence / 0187d8f7-4f52-461a-af17-77a5ea7bc7f0

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous