rhosus

1
/*
2
 * Copyright (c) 2022.
3
 * Licensed to the Parasource Foundation under one or more contributor license agreements.  See the NOTICE file distributed with this work for additional information regarding copyright ownership.  The Parasource licenses this file to you under the Parasource License, Version 2.0 (the "License"); you may not use this file except in compliance with the License.
4
 * You may obtain a copy of the License at http://www.parasource.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.
5
 * See the License for the specific language governing permissions and limitations under the License.
6
 */
7

8
package main
9

10
import (
11
	"fmt"
12
	"github.com/parasource/rhosus/rhosus/api"
13
	"github.com/parasource/rhosus/rhosus/auth"
14
	"github.com/parasource/rhosus/rhosus/registry"
15
	"github.com/parasource/rhosus/rhosus/registry/cluster"
16
	"github.com/parasource/rhosus/rhosus/registry/storage"
17
	"github.com/parasource/rhosus/rhosus/util"
18
	"github.com/parasource/rhosus/rhosus/util/uuid"
19
	"github.com/rs/zerolog/log"
20
	"github.com/spf13/cobra"
21
	"github.com/spf13/pflag"
22
	"github.com/spf13/viper"
23
	"io"
24
	"os"
25
	"os/signal"
26
	"path"
27
	"runtime"
28
	"syscall"
29
	"time"
30
)
31

32
const (
33
	uuidFileName = "registry.uuid"
34
)
35

36
var configDefaults = map[string]interface{}{
37
	"gomaxprocs": 0,
38
	// http file server host and port
39
	"api_addr":     "127.0.0.1:8000",
40
	"cluster_addr": "127.0.0.1:8100",
41
	"etcd_addr":    "127.0.0.1:2379",
42
	"rhosus_path":  "/var/lib/rhosus",
43

44
	// path for wal
45
	"wal_path": "wal",
46
	// seconds to wait til force shutdown
47
	"shutdown_timeout": 30,
48
	// how many times a file should be replicated
49
	"replication_factor": 1,
50
	// block size in bytes
51
	"block_size": 4096,
52
}
53

54
type DefaultChecker struct {
55
	flags *pflag.FlagSet
56
}
57

58
func (c *DefaultChecker) checkIfUsingDefault(name string) bool {
59
	flag := true
60

61
	flag = flag && os.Getenv(name) == ""
62
	//flag = flag && c.flags.Lookup(name) == nil
63

64
	return flag
65
}
66

67
var checker *DefaultChecker
68

69
func init() {
70
	rootCmd.Flags().String("api_addr", "127.0.0.1:8000", "api server address")
71
	rootCmd.Flags().String("cluster_addr", "127.0.0.1:8100", "cluster server address")
72
	rootCmd.Flags().String("etcd_addr", "127.0.0.1:2379", "etcd service discovery address")
73
	rootCmd.Flags().String("rhosus_path", "/var/lib/rhosus", "rhosus data path")
74
	rootCmd.Flags().Int("shutdown_timeout", 30, "node graceful shutdown timeout")
75
	rootCmd.Flags().Int("replication_factor", 30, "replication factor")
76
	rootCmd.Flags().Int("block_size", 4096, "block size in bytes")
77

78
	viper.BindPFlag("api_addr", rootCmd.Flags().Lookup("cluster_addr"))
79
	viper.BindPFlag("cluster_addr", rootCmd.Flags().Lookup("cluster_addr"))
80
	viper.BindPFlag("etcd_addr", rootCmd.Flags().Lookup("etcd_addr"))
81
	viper.BindPFlag("rhosus_path", rootCmd.Flags().Lookup("rhosus_path"))
82
	viper.BindPFlag("shutdown_timeout", rootCmd.Flags().Lookup("shutdown_timeout"))
83
	viper.BindPFlag("replication_factor", rootCmd.Flags().Lookup("replication_factor"))
84
	viper.BindPFlag("block_size", rootCmd.Flags().Lookup("block size in bytes"))
85

86
	checker = &DefaultChecker{
87
		flags: rootCmd.Flags(),
88
	}
89
}
90

91
var rootCmd = &cobra.Command{
92
	Use: "rhosusr",
93
	Run: func(cmd *cobra.Command, args []string) {
94

95
		printWelcome()
96

97
		for k, v := range configDefaults {
98
			viper.SetDefault(k, v)
99
		}
100

101
		bindEnvs := []string{
102
			"api_addr", "cluster_addr", "etcd_addr", "rhosus_path",
103
			"shutdown_timeout", "replication_factor", "block_size",
104
		}
105
		for _, env := range bindEnvs {
106
			err := viper.BindEnv(env)
107
			if err != nil {
108
				log.Fatal().Err(err).Msg("error binding env variable")
109
			}
110
		}
111

112
		if os.Getenv("GOMAXPROCS") == "" {
113
			if viper.IsSet("gomaxprocs") && viper.GetInt("gomaxprocs") > 0 {
114
				runtime.GOMAXPROCS(viper.GetInt("gomaxprocs"))
115
			} else {
116
				runtime.GOMAXPROCS(runtime.NumCPU())
117
			}
118
		}
119

120
		v := viper.GetViper()
121

122
		shutdownCh := make(chan struct{}, 1)
123

124
		rhosusPath := v.GetString("rhosus_path")
125

126
		// If rhosus home directory does not exist already,
127
		// we create a new one
128
		if ok := util.FileExists(rhosusPath); !ok {
129
			err := os.Mkdir(rhosusPath, 0755)
130
			if err != nil {
131
				log.Fatal().Err(err).Msg("error creating rhosus home directory")
132
			}
133
		}
134
		if err := util.TestDirWritable(rhosusPath); err != nil {
135
			log.Fatal().Err(err).Msg("rhosus home directory is not writable")
136
		}
137

138
		s, err := storage.NewStorage(storage.Config{
139
			Path:          path.Join(rhosusPath, "registry"),
140
			WriteTimeoutS: 10,
141
			NumWorkers:    5,
142
		})
143
		if err != nil {
144
			log.Fatal().Err(err).Msg("error creating storage")
145
		}
146
		roleManager, err := auth.NewRoleManager(s)
147
		if err != nil {
148
			log.Fatal().Err(err).Msg("error creating role manager")
149
		}
150
		tokenManager, err := auth.NewTokenStore(s)
151
		if err != nil {
152
			log.Fatal().Err(err).Msg("error creating token manager")
153
		}
154

155
		authMethods := map[string]auth.Authenticator{
156
			"credentials": auth.NewCredentialsAuth(roleManager, tokenManager),
157
		}
158

159
		conf, err := registryConfig(v, s)
160

161
		r, err := registry.NewRegistry(conf)
162
		if err != nil {
163
			log.Fatal().Err(err).Msg("error creating registry instance")
164
		}
165

166
		go r.Start()
167

168
		apiAddr := v.GetString("api_addr")
169
		httpApi, err := api.NewApi(r, tokenManager, api.Config{
170
			Address:     apiAddr,
171
			AuthMethods: authMethods,
172
		})
173
		go httpApi.Run()
174

175
		go handleSignals(shutdownCh)
176
		for {
177
			select {
178
			case <-shutdownCh:
179
				httpApi.Shutdown()
180
				r.Shutdown()
181
				return
182
			}
183
		}
184
	},
185
}
186

187
func registryConfig(v *viper.Viper, s *storage.Storage) (registry.Config, error) {
188
	if os.Getenv("API_ADDR") == "" {
189
		log.Warn().Msg("API_ADDR is not set explicitly, falling back to default")
190
	}
191
	if os.Getenv("CLUSTER_ADDR") == "" {
192
		log.Warn().Msg("CLUSTER_ADDR is not set explicitly, falling back to default")
193
	}
194

195
	clusterAddr := v.GetString("cluster_addr")
196
	rhosusPath := v.GetString("rhosus_path")
197
	etcdAddr := v.GetString("etcd_addr")
198

199
	// Generating id for Registry
200
	registryId := getId(rhosusPath, true)
201

202
	conf := registry.Config{
203
		ID:         registryId,
204
		RhosusPath: rhosusPath,
205
		EtcdAddr:   etcdAddr,
206
		Storage:    s,
207
		Cluster: cluster.Config{
208
			WalPath:     path.Join(rhosusPath, "wal"),
209
			ClusterAddr: clusterAddr,
210
			ID:          registryId,
211
		},
212
	}
213

214
	return conf, nil
215
}
216

217
func getId(rhosusPath string, persistent bool) string {
218
	var id string
219

220
	if !persistent {
221
		v4id, _ := uuid.NewV4()
222
		return v4id.String()
223
	}
224

225
	uuidFilePath := path.Join(rhosusPath, uuidFileName)
226

227
	// since we are just testing, we don't need that yet
228
	if util.FileExists(uuidFilePath) {
229
		file, err := os.OpenFile(uuidFilePath, os.O_RDONLY, 0666)
230
		defer file.Close()
231

232
		if err != nil {
233
			log.Fatal().Err(err).Str("path", uuidFilePath).Msg("error opening node uuid file")
234
		}
235
		data, err := io.ReadAll(file)
236
		if err != nil {
237
			log.Fatal().Err(err).Str("path", uuidFilePath).Msg("error reading node uuid file")
238
		}
239

240
		id = string(data)
241
	} else {
242
		v4uid, _ := uuid.NewV4()
243
		id = v4uid.String()
244

245
		file, err := os.OpenFile(uuidFilePath, os.O_CREATE|os.O_RDWR, 0755)
246
		if err != nil {
247
			log.Fatal().Err(err).Msg("error opening node uuid file")
248
		}
249
		defer file.Close()
250
		file.Write([]byte(id))
251
	}
252

253
	return id
254
}
255

256
func handleSignals(shutdownCh chan<- struct{}) {
257
	sigc := make(chan os.Signal, 1)
258
	signal.Notify(sigc, syscall.SIGHUP, syscall.SIGINT, os.Interrupt, syscall.SIGTERM)
259
	for {
260
		sig := <-sigc
261
		log.Info().Str("signal", sig.String()).Msg("system signal received")
262
		switch sig {
263
		case syscall.SIGHUP:
264

265
		case syscall.SIGINT, os.Interrupt, syscall.SIGTERM:
266

267
			pidFile := viper.GetString("pid_file")
268
			shutdownTimeout := time.Duration(viper.GetInt("shutdown_timeout")) * time.Second
269

270
			close(shutdownCh)
271

272
			go time.AfterFunc(shutdownTimeout, func() {
273
				if pidFile != "" {
274
					os.Remove(pidFile)
275
				}
276
				os.Exit(1)
277
			})
278
		}
279
	}
280
}
281

282
func printWelcome() {
283
	welcome := "    ____  __  ______  _____ __  _______\n   / __ \\/ / / / __ \\/ ___// / / / ___/\n  / /_/ / /_/ / / / /\\__ \\/ / / /\\__ \\ \n / _, _/ __  / /_/ /___/ / /_/ /___/ / \n/_/ |_/_/ /_/\\____//____/\\____//____/  \n                                       "
284
	fmt.Println(welcome)
285

286
	fmt.Println("\n|------ Rhosus registry")
287
	fmt.Println("|------ Version " + util.Version() + "\n")
288
}
289