Vihren Atlas

lite_server.go ×25

Code frontier

unlabeled · c_c479527d8a3b

7 tests · 21045 LOC · 601 files · introduces 0 tests · 303 LOC · 8 files

Introduces — evidence that enters the hierarchy at this concept

Code
51 ranges303 lines · 8 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
4068 ranges21045 lines · 601 files · Browse complete extent
All tests (intent)
7 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

8 files ranked by introduced lines: 303 introduced LOC across 51 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/temporaltest/internal/lite_server.go 175 introduced LOC · 25 ranges

Open complete file

72 }
73
74 > func (cfg *LiteServerConfig) apply(serverConfig *config.Config) { lite_server.go
75 > sqliteConfig := config.SQL{
76 > PluginName: sqliteplugin.PluginName,
77 > ConnectAttributes: make(map[string]string),
78 > DatabaseName: cfg.DatabaseFilePath,
79 > }
80 > if cfg.Ephemeral {
81 > sqliteConfig.ConnectAttributes["mode"] = "memory"
82 > sqliteConfig.ConnectAttributes["cache"] = "shared"
83 > // TODO(jlegrone): investigate whether a randomized db name is necessary when running in shared cache mode:
84 > // https://www.sqlite.org/sharedcache.html
85 > sqliteConfig.DatabaseName = fmt.Sprintf("%d", rand.Intn(9999999))
86 > } else {
87 sqliteConfig.ConnectAttributes["mode"] = "rwc"
88 }
89
90 > for k, v := range cfg.SQLitePragmas { lite_server.go
91 sqliteConfig.ConnectAttributes["_"+k] = v
92 }
93
94 > if cfg.FrontendPort == 0 { lite_server.go
95 > cfg.FrontendPort = freeport.MustGetFreePort()
96 > }
97 > if cfg.MetricsPort == 0 {
98 > cfg.MetricsPort = freeport.MustGetFreePort()
99 > }
100 > pprofPort := freeport.MustGetFreePort()
101 >
102 > serverConfig.Global.Membership = config.Membership{
103 > MaxJoinDuration: 30 * time.Second,
104 > BroadcastAddress: localBroadcastAddress,
105 > }
106 > serverConfig.Global.Metrics = &metrics.Config{
107 > Prometheus: &metrics.PrometheusConfig{
108 > ListenAddress: fmt.Sprintf("%s:%d", cfg.FrontendIP, cfg.MetricsPort),
109 > HandlerPath: "/metrics",
110 > },
111 > }
112 > serverConfig.Global.PProf = config.PProf{Port: pprofPort}
113 > serverConfig.Persistence = config.Persistence{
114 > DefaultStore: sqliteplugin.PluginName,
115 > VisibilityStore: sqliteplugin.PluginName,
116 > NumHistoryShards: 1,
117 > DataStores: map[string]config.DataStore{
118 > sqliteplugin.PluginName: {SQL: &sqliteConfig},
119 > },
120 > }
121 > serverConfig.ClusterMetadata = &cluster.Config{
122 > EnableGlobalNamespace: false,
123 > FailoverVersionIncrement: 10,
124 > MasterClusterName: "active",
125 > CurrentClusterName: "active",
126 > ClusterInformation: map[string]cluster.ClusterInformation{
127 > "active": {
128 > Enabled: true,
129 > InitialFailoverVersion: 1,
130 > RPCAddress: fmt.Sprintf("%s:%d", localBroadcastAddress, cfg.FrontendPort),
131 > },
132 > },
133 > }
134 > serverConfig.DCRedirectionPolicy = config.DCRedirectionPolicy{
135 > Policy: "noop",
136 > }
137 > serverConfig.Services = map[string]config.Service{
138 > "frontend": cfg.mustGetService(0),
139 > "history": cfg.mustGetService(1),
140 > "matching": cfg.mustGetService(2),
141 > "worker": cfg.mustGetService(3),
142 > }
143 > serverConfig.Archival = config.Archival{
144 > History: config.HistoryArchival{
145 > State: "disabled",
146 > EnableRead: false,
147 > Provider: nil,
148 > },
149 > Visibility: config.VisibilityArchival{
150 > State: "disabled",
151 > EnableRead: false,
152 > Provider: nil,
153 > },
154 > }
155 > // TODO(dnr): Figure out why server fails to start when PublicClient is not set with error:
156 > // panic: Client must be created with client.Dial() or client.NewLazyClient()
157 > // See also: https://github.com/temporalio/temporal/pull/4026#discussion_r1149808018
158 > serverConfig.PublicClient = config.PublicClient{
159 > HostPort: fmt.Sprintf("%s:%d", localBroadcastAddress, cfg.FrontendPort),
160 > }
161 > serverConfig.NamespaceDefaults = config.NamespaceDefaults{
162 > Archival: config.ArchivalNamespaceDefaults{
163 > History: config.HistoryArchivalNamespaceDefaults{
164 > State: "disabled",
165 > },
166 > Visibility: config.VisibilityArchivalNamespaceDefaults{
167 > State: "disabled",
168 > },
169 > },
170 > }
171 }
172
173 > func (cfg *LiteServerConfig) applyDefaults() { lite_server.go
174 > if cfg.BaseConfig == nil {
175 > cfg.BaseConfig = &config.Config{}
176 > }
177 > if cfg.Logger == nil {
178 cfg.Logger = log.NewZapLogger(log.BuildZapLogger(log.Config{
179 Stdout: true,
184 }
185
186 > func (cfg *LiteServerConfig) validate() error { lite_server.go
187 > for pragma := range cfg.SQLitePragmas {
188 if _, ok := supportedPragmas[strings.ToLower(pragma)]; !ok {
189 return fmt.Errorf("unsupported SQLite pragma %q. allowed pragmas: %v", pragma, getAllowedPragmas())
191 }
192
193 > if cfg.Ephemeral && cfg.DatabaseFilePath != "" { lite_server.go
194 return fmt.Errorf("config option DatabaseFilePath is not supported in ephemeral mode")
195 }
196 > if !cfg.Ephemeral && cfg.DatabaseFilePath == "" { lite_server.go
197 return fmt.Errorf("config option DatabaseFilePath is required when ephemeral mode disabled")
198 }
199
200 > return nil lite_server.go
201 }
202
214 // Always use BaseConfig instead of the WithConfig server option, as WithConfig overrides all
215 // LiteServer specific settings.
216 > func NewLiteServer(liteConfig *LiteServerConfig, opts ...temporal.ServerOption) (*LiteServer, error) { lite_server.go
217 > liteConfig.applyDefaults()
218 > if err := liteConfig.validate(); err != nil {
219 return nil, err
220 }
221
222 > liteConfig.apply(liteConfig.BaseConfig) lite_server.go
223 >
224 > sqlConfig := liteConfig.BaseConfig.Persistence.DataStores[sqliteplugin.PluginName].SQL
225 >
226 > if !liteConfig.Ephemeral {
227 // Apply migrations if file does not already exist
228 if _, err := os.Stat(liteConfig.DatabaseFilePath); os.IsNotExist(err) {
240
241 // Pre-create namespaces
242 > var namespaces []*sqlite.NamespaceConfig lite_server.go
243 > for _, ns := range liteConfig.Namespaces {
244 > nsConfig, err := sqlite.NewNamespaceConfig(
245 > liteConfig.BaseConfig.ClusterMetadata.CurrentClusterName,
246 > ns,
247 > false,
248 > liteConfig.SearchAttributes,
249 > )
250 > if err != nil {
251 return nil, fmt.Errorf("error creating namespace config: %w", err)
252 }
253 > namespaces = append(namespaces, nsConfig) lite_server.go
254 }
255 > if err := sqlite.CreateNamespaces(sqlConfig, namespaces...); err != nil { lite_server.go
256 return nil, fmt.Errorf("error creating namespaces: %w", err)
257 }
258
259 > authorizer, err := authorization.GetAuthorizerFromConfig(&liteConfig.BaseConfig.Global.Authorization) lite_server.go
260 > if err != nil {
261 return nil, fmt.Errorf("unable to instantiate authorizer: %w", err)
262 }
263
264 > claimMapper, err := authorization.GetClaimMapperFromConfig(&liteConfig.BaseConfig.Global.Authorization, liteConfig.Logger) lite_server.go
265 > if err != nil {
266 return nil, fmt.Errorf("unable to instantiate claim mapper: %w", err)
267 }
268
269 > serverOpts := []temporal.ServerOption{ lite_server.go
270 > temporal.WithConfig(liteConfig.BaseConfig),
271 > temporal.ForServices(temporal.DefaultServices),
272 > temporal.WithLogger(liteConfig.Logger),
273 > temporal.WithAuthorizer(authorizer),
274 > temporal.WithClaimMapper(func(cfg *config.Config) authorization.ClaimMapper {
275 > return claimMapper
276 > }),
277 }
278
279 > if len(liteConfig.DynamicConfig) > 0 { lite_server.go
280 > // To prevent having to code fall-through semantics right now, we currently
281 > // eagerly fail if dynamic config is being configured in two ways
282 > if liteConfig.BaseConfig.DynamicConfigClient != nil {
283 return nil, fmt.Errorf("unable to have file-based dynamic config and individual dynamic config values")
284 }
285 > serverOpts = append(serverOpts, temporal.WithDynamicConfigClient(liteConfig.DynamicConfig)) lite_server.go
286 }
287
288 // Apply options from arguments
289 > serverOpts = append(serverOpts, opts...) lite_server.go
290 >
291 > srv, err := temporal.NewServer(serverOpts...)
292 > if err != nil {
293 return nil, fmt.Errorf("unable to instantiate server: %w", err)
294 }
295
296 > s := &LiteServer{ lite_server.go
297 > internal: srv,
298 > frontendHostPort: liteConfig.BaseConfig.PublicClient.HostPort,
299 > }
300 >
301 > return s, nil
302 }
303
304 // Start temporal server.
305 > func (s *LiteServer) Start() error { lite_server.go
306 > // We wrap Server instead of simply embedding it in the LiteServer struct so
307 > // that it's possible to add additional lifecycle hooks here if necessary.
308 > return s.internal.Start()
309 > }
310
311 // Stop the server.
327 //
328 // Note that options.HostPort will always be overridden.
329 > func (s *LiteServer) NewClientWithOptions(ctx context.Context, options client.Options) (client.Client, error) { lite_server.go
330 > options.HostPort = s.frontendHostPort
331 > return client.Dial(options)
332 > }
333
334 // FrontendHostPort returns the host:port for this server.
351 }
352
353 > func (cfg *LiteServerConfig) mustGetService(frontendPortOffset int) config.Service { lite_server.go
354 > svc := config.Service{
355 > RPC: config.RPC{
356 > GRPCPort: cfg.FrontendPort + frontendPortOffset,
357 > MembershipPort: freeport.MustGetFreePort(),
358 > BindOnLocalHost: true,
359 > BindOnIP: "",
360 > },
361 > }
362 >
363 > // Assign any open port when configured to use dynamic ports
364 > if frontendPortOffset != 0 {
365 > svc.RPC.GRPCPort = freeport.MustGetFreePort()
366 > }
367
368 // Optionally bind frontend to IPv4 address
369 > if frontendPortOffset == 0 && cfg.FrontendIP != "" { lite_server.go
370 > svc.RPC.BindOnLocalHost = false
371 > svc.RPC.BindOnIP = cfg.FrontendIP
372 > }
373
374 > return svc lite_server.go
375 }
go.temporal.io/server/schema/sqlite/setup.go 56 introduced LOC · 10 ranges

Open complete file

91 //
92 // Note: this function may receive breaking changes or be removed in the future.
93 > func CreateNamespaces(cfg *config.SQL, namespaces ...*NamespaceConfig) error { setup.go
94 > db, err := sql.NewSQLDB(sqlplugin.DbKindUnknown, cfg, resolver.NewNoopResolver(), log.NewNoopLogger(), metrics.NoopMetricsHandler)
95 > if err != nil {
96 return fmt.Errorf("unable to create SQLite admin DB: %w", err)
97 }
98 > defer func() { _ = db.Close() }() setup.go
99
100 > for _, ns := range namespaces { setup.go
101 > if err := createNamespaceIfNotExists(db, ns); err != nil {
102 return fmt.Errorf("error creating namespace %q: %w", ns.Detail.Info.Name, err)
103 }
104 }
105
106 > return nil setup.go
107 }
108
116 global bool,
117 customSearchAttributes map[string]enumspb.IndexedValueType,
118 > ) (*NamespaceConfig, error) { setup.go
119 > dbCustomSearchAttributes := sadefs.GetDBIndexSearchAttributes(nil).CustomSearchAttributes
120 > fieldToAliasMap := map[string]string{}
121 > for saName, saType := range customSearchAttributes {
122 var targetFieldName string
123 var cntUsed int
142 }
143
144 > detail := persistencespb.NamespaceDetail{ setup.go
145 > Info: &persistencespb.NamespaceInfo{
146 > Id: primitives.NewUUID().String(),
147 > State: enumspb.NAMESPACE_STATE_REGISTERED,
148 > Name: namespace,
149 > },
150 > Config: &persistencespb.NamespaceConfig{
151 > Retention: timestamp.DurationFromHours(24),
152 > HistoryArchivalState: enumspb.ARCHIVAL_STATE_DISABLED,
153 > VisibilityArchivalState: enumspb.ARCHIVAL_STATE_DISABLED,
154 > CustomSearchAttributeAliases: fieldToAliasMap,
155 > },
156 > ReplicationConfig: &persistencespb.NamespaceReplicationConfig{
157 > ActiveClusterName: activeClusterName,
158 > Clusters: []string{activeClusterName},
159 > },
160 > FailoverVersion: common.EmptyVersion,
161 > FailoverNotificationVersion: -1,
162 > }
163 > return &NamespaceConfig{
164 > Detail: &detail,
165 > IsGlobal: global,
166 > }, nil
167 }
168
169 > func createNamespaceIfNotExists(db sqlplugin.DB, namespace *NamespaceConfig) error { setup.go
170 > var (
171 > name = namespace.Detail.GetInfo().GetName()
172 > id = primitives.MustParseUUID(namespace.Detail.GetInfo().GetId())
173 > )
174 >
175 > // Return early if namespace already exists
176 > rows, err := db.SelectFromNamespace(context.Background(), sqlplugin.NamespaceFilter{
177 > Name: &name,
178 > })
179 > if err == nil && len(rows) > 0 {
180 return nil
181 }
182
183 > blob, err := serialization.NewSerializer().NamespaceDetailToBlob(namespace.Detail) setup.go
184 > if err != nil {
185 return err
186 }
187
188 > if _, err := db.InsertIntoNamespace(context.Background(), &sqlplugin.NamespaceRow{ setup.go
189 > ID: id,
190 > Name: name,
191 > Data: blob.GetData(),
192 > DataEncoding: blob.GetEncodingType().String(),
193 > IsGlobal: namespace.IsGlobal,
194 > NotificationVersion: 0,
195 > }); err != nil {
196 return err
197 }
198
199 > return nil setup.go
200 }
go.temporal.io/server/temporaltest/server.go 50 introduced LOC · 9 ranges

Open complete file

66 //
67 // It is configured to use a pre-registered test namespace and will be closed on TestServer.Stop.
68 > func (ts *TestServer) GetDefaultClient() client.Client { server.go
69 > if ts.defaultClient == nil {
70 > ts.defaultClient = ts.NewClientWithOptions(ts.defaultClientOptions)
71 > }
72 > return ts.defaultClient
73 }
74
90 // If no namespace option is set it will use a pre-registered test namespace.
91 // The returned client will be closed on TestServer.Stop.
92 > func (ts *TestServer) NewClientWithOptions(opts client.Options) client.Client { server.go
93 > if opts.Namespace == "" {
94 > opts.Namespace = ts.defaultTestNamespace
95 > }
96 > if opts.Logger == nil {
97 > opts.Logger = &testLogger{ts.t}
98 > }
99
100 > ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) server.go
101 > defer cancel()
102 >
103 > c, err := ts.server.NewClientWithOptions(ctx, opts)
104 > if err != nil {
105 ts.fatal(fmt.Errorf("error creating client: %w", err))
106 }
107
108 > ts.clients = append(ts.clients, c) server.go
109 >
110 > return c
111 }
112
130 // If not specifying the WithT option, the caller should execute Stop when finished to close
131 // the server and release resources.
132 > func NewServer(opts ...TestServerOption) *TestServer { server.go
133 > testNamespace := fmt.Sprintf("temporaltest-%d", rand.Intn(1e6))
134 >
135 > ts := TestServer{
136 > defaultTestNamespace: testNamespace,
137 > }
138 >
139 > // Apply options
140 > for _, opt := range opts {
141 > opt.apply(&ts)
142 > }
143
144 > if ts.t != nil { server.go
145 ts.t.Cleanup(ts.Stop)
146 }
147
148 > s, err := temporalite.NewLiteServer(&temporalite.LiteServerConfig{ server.go
149 > Namespaces: []string{ts.defaultTestNamespace},
150 > Ephemeral: true,
151 > Logger: log.NewNoopLogger(),
152 > DynamicConfig: dynamicconfig.StaticClient{
153 > dynamicconfig.ForceSearchAttributesCacheRefreshOnRead.Key(): []dynamicconfig.ConstrainedValue{{Value: true}},
154 > },
155 > // Disable "accept incoming network connections?" prompt on macOS
156 > FrontendIP: "127.0.0.1",
157 > }, ts.serverOptions...)
158 > if err != nil {
159 ts.fatal(fmt.Errorf("error creating server: %w", err))
160 }
161 > ts.server = s server.go
162 >
163 > // Start does not block as long as InterruptOn is unset.
164 > if err := s.Start(); err != nil {
165 ts.fatal(err)
166 }
167
168 // This sleep helps avoid a panic in github.com/temporalio/[email protected]/swim/labels.go:175
169 > time.Sleep(100 * time.Millisecond) server.go
170 >
171 > return &ts
172 }
go.temporal.io/server/temporal/server_option.go 8 introduced LOC · 2 ranges

Open complete file

96
97 // WithAuthorizer sets a low level authorizer to allow/deny all API calls
98 > func WithAuthorizer(authorizer authorization.Authorizer) ServerOption { server_option.go
99 > return applyFunc(func(s *serverOptions) {
100 > s.authorizer = authorizer
101 > })
102 }
103
110
111 // WithClaimMapper configures a role mapper for authorization
112 > func WithClaimMapper(claimMapper func(cfg *config.Config) authorization.ClaimMapper) ServerOption { server_option.go
113 > return applyFunc(func(s *serverOptions) {
114 > s.claimMapper = claimMapper(s.config)
115 > })
116 }
117
go.temporal.io/server/temporaltest/options.go 5 introduced LOC · 2 ranges

Open complete file

15 type applyFunc func(*TestServer)
16
17 > func (f applyFunc) apply(s *TestServer) { f(s) } options.go
18
19 // WithT directs all worker and client logs to the test logger.
21 // If this option is specified, then server will automatically be stopped when the
22 // test completes.
23 > func WithT(t *testing.T) TestServerOption { options.go
24 > return applyFunc(func(server *TestServer) {
25 > server.t = t
26 > })
27 }
28
go.temporal.io/server/service/frontend/service.go 4 introduced LOC · 1 range

Open complete file

510 }
511 }()
512 > } else { service.go
513 > s.logger.Warn("HTTP API port has not been set. Nexus HTTP endpoints will not be available. " +
514 > "To enable Nexus, follow these instructions: https://github.com/temporalio/temporal/blob/main/docs/architecture/nexus.md#enabling-nexus.")
515 > }
516
517 go s.membershipMonitor.Start()
go.temporal.io/server/common/primitives/timestamp/duration.go 3 introduced LOC · 1 range

Open complete file

43 }
44
45 > func DurationFromHours(h int64) *durationpb.Duration { duration.go
46 > return durationMultipleOf(h, time.Hour)
47 > }
48
49 func DurationFromDays(d int32) *durationpb.Duration {
go.temporal.io/server/service/frontend/fx.go 2 introduced LOC · 1 range

Open complete file

1018 ) (*HTTPAPIServer, error) {
1019 if !httpEnabled(cfg, serviceName) {
1020 > return nil, nil fx.go
1021 > }
1022 rpcConfig := cfg.Services[string(serviceName)].RPC
1023 return NewHTTPAPIServer(

Direct hierarchy

Parents

Children