mirror of
https://github.com/gogrlx/nats-server.git
synced 2026-04-02 03:38:42 -07:00
367 lines
8.8 KiB
Go
367 lines
8.8 KiB
Go
// Copyright 2012-2013 Apcera Inc. All rights reserved.
|
|
|
|
package server
|
|
|
|
import (
|
|
"fmt"
|
|
"io/ioutil"
|
|
"net"
|
|
"net/url"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/gnatsd/conf"
|
|
)
|
|
|
|
// Options block for gnatsd server.
|
|
type Options struct {
|
|
Host string `json:"addr"`
|
|
Port int `json:"port"`
|
|
Trace bool `json:"-"`
|
|
Debug bool `json:"-"`
|
|
NoLog bool `json:"-"`
|
|
NoSigs bool `json:"-"`
|
|
Logtime bool `json:"-"`
|
|
MaxConn int `json:"max_connections"`
|
|
Username string `json:"user,omitempty"`
|
|
Password string `json:"-"`
|
|
Authorization string `json:"-"`
|
|
PingInterval time.Duration `json:"ping_interval"`
|
|
MaxPingsOut int `json:"ping_max"`
|
|
HTTPPort int `json:"http_port"`
|
|
SslTimeout float64 `json:"ssl_timeout"`
|
|
AuthTimeout float64 `json:"auth_timeout"`
|
|
MaxControlLine int `json:"max_control_line"`
|
|
MaxPayload int `json:"max_payload"`
|
|
MaxPending int `json:"max_pending_size"`
|
|
ClusterHost string `json:"addr"`
|
|
ClusterPort int `json:"port"`
|
|
ClusterUsername string `json:"-"`
|
|
ClusterPassword string `json:"-"`
|
|
ClusterAuthTimeout float64 `json:"auth_timeout"`
|
|
ProfPort int `json:"-"`
|
|
PidFile string `json:"-"`
|
|
LogFile string `json:"-"`
|
|
Syslog bool `json:"-"`
|
|
RemoteSyslog string `json:"-"`
|
|
Routes []*url.URL `json:"-"`
|
|
RoutesStr string `json:"-"`
|
|
}
|
|
|
|
type authorization struct {
|
|
user string
|
|
pass string
|
|
timeout float64
|
|
}
|
|
|
|
// ProcessConfigFile processes a configuration file.
|
|
// FIXME(dlc): Hacky
|
|
func ProcessConfigFile(configFile string) (*Options, error) {
|
|
opts := &Options{}
|
|
|
|
if configFile == "" {
|
|
return opts, nil
|
|
}
|
|
|
|
data, err := ioutil.ReadFile(configFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error opening config file: %v", err)
|
|
}
|
|
|
|
m, err := conf.Parse(string(data))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for k, v := range m {
|
|
switch strings.ToLower(k) {
|
|
case "port":
|
|
opts.Port = int(v.(int64))
|
|
case "host", "net":
|
|
opts.Host = v.(string)
|
|
case "debug":
|
|
opts.Debug = v.(bool)
|
|
case "trace":
|
|
opts.Trace = v.(bool)
|
|
case "logtime":
|
|
opts.Logtime = v.(bool)
|
|
case "authorization":
|
|
am := v.(map[string]interface{})
|
|
auth := parseAuthorization(am)
|
|
opts.Username = auth.user
|
|
opts.Password = auth.pass
|
|
opts.AuthTimeout = auth.timeout
|
|
case "http_port", "monitor_port":
|
|
opts.HTTPPort = int(v.(int64))
|
|
case "cluster":
|
|
cm := v.(map[string]interface{})
|
|
if err := parseCluster(cm, opts); err != nil {
|
|
return nil, err
|
|
}
|
|
case "logfile", "log_file":
|
|
opts.LogFile = v.(string)
|
|
case "syslog":
|
|
opts.Syslog = v.(bool)
|
|
case "remote_syslog":
|
|
opts.RemoteSyslog = v.(string)
|
|
case "pidfile", "pid_file":
|
|
opts.PidFile = v.(string)
|
|
case "prof_port":
|
|
opts.ProfPort = int(v.(int64))
|
|
case "max_control_line":
|
|
opts.MaxControlLine = int(v.(int64))
|
|
case "max_payload":
|
|
opts.MaxPayload = int(v.(int64))
|
|
case "max_pending_size", "max_pending":
|
|
opts.MaxPending = int(v.(int64))
|
|
case "max_connections", "max_conn":
|
|
opts.MaxConn = int(v.(int64))
|
|
}
|
|
}
|
|
return opts, nil
|
|
}
|
|
|
|
// parseCluster will parse the cluster config.
|
|
func parseCluster(cm map[string]interface{}, opts *Options) error {
|
|
for mk, mv := range cm {
|
|
switch strings.ToLower(mk) {
|
|
case "port":
|
|
opts.ClusterPort = int(mv.(int64))
|
|
case "host", "net":
|
|
opts.ClusterHost = mv.(string)
|
|
case "authorization":
|
|
am := mv.(map[string]interface{})
|
|
auth := parseAuthorization(am)
|
|
opts.ClusterUsername = auth.user
|
|
opts.ClusterPassword = auth.pass
|
|
opts.ClusterAuthTimeout = auth.timeout
|
|
case "routes":
|
|
ra := mv.([]interface{})
|
|
opts.Routes = make([]*url.URL, 0, len(ra))
|
|
for _, r := range ra {
|
|
routeURL := r.(string)
|
|
url, err := url.Parse(routeURL)
|
|
if err != nil {
|
|
return fmt.Errorf("error parsing route url [%q]", routeURL)
|
|
}
|
|
opts.Routes = append(opts.Routes, url)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Helper function to parse Authorization configs.
|
|
func parseAuthorization(am map[string]interface{}) authorization {
|
|
auth := authorization{}
|
|
for mk, mv := range am {
|
|
switch strings.ToLower(mk) {
|
|
case "user", "username":
|
|
auth.user = mv.(string)
|
|
case "pass", "password":
|
|
auth.pass = mv.(string)
|
|
case "timeout":
|
|
at := float64(1)
|
|
switch mv.(type) {
|
|
case int64:
|
|
at = float64(mv.(int64))
|
|
case float64:
|
|
at = mv.(float64)
|
|
}
|
|
auth.timeout = at
|
|
}
|
|
}
|
|
return auth
|
|
}
|
|
|
|
// MergeOptions will merge two options giving preference to the flagOpts
|
|
// if the item is present.
|
|
func MergeOptions(fileOpts, flagOpts *Options) *Options {
|
|
if fileOpts == nil {
|
|
return flagOpts
|
|
}
|
|
if flagOpts == nil {
|
|
return fileOpts
|
|
}
|
|
// Merge the two, flagOpts override
|
|
opts := *fileOpts
|
|
|
|
if flagOpts.Port != 0 {
|
|
opts.Port = flagOpts.Port
|
|
}
|
|
if flagOpts.Host != "" {
|
|
opts.Host = flagOpts.Host
|
|
}
|
|
if flagOpts.Username != "" {
|
|
opts.Username = flagOpts.Username
|
|
}
|
|
if flagOpts.Password != "" {
|
|
opts.Password = flagOpts.Password
|
|
}
|
|
if flagOpts.Authorization != "" {
|
|
opts.Authorization = flagOpts.Authorization
|
|
}
|
|
if flagOpts.HTTPPort != 0 {
|
|
opts.HTTPPort = flagOpts.HTTPPort
|
|
}
|
|
if flagOpts.Debug {
|
|
opts.Debug = true
|
|
}
|
|
if flagOpts.Trace {
|
|
opts.Trace = true
|
|
}
|
|
if flagOpts.Logtime {
|
|
opts.Logtime = true
|
|
}
|
|
if flagOpts.LogFile != "" {
|
|
opts.LogFile = flagOpts.LogFile
|
|
}
|
|
if flagOpts.PidFile != "" {
|
|
opts.PidFile = flagOpts.PidFile
|
|
}
|
|
if flagOpts.ProfPort != 0 {
|
|
opts.ProfPort = flagOpts.ProfPort
|
|
}
|
|
if flagOpts.RoutesStr != "" {
|
|
mergeRoutes(&opts, flagOpts)
|
|
}
|
|
return &opts
|
|
}
|
|
|
|
func RoutesFromStr(routesStr string) []*url.URL {
|
|
routes := strings.Split(routesStr, ",")
|
|
if len(routes) == 0 {
|
|
return nil
|
|
}
|
|
routeUrls := []*url.URL{}
|
|
for _, r := range routes {
|
|
r = strings.TrimSpace(r)
|
|
u, _ := url.Parse(r)
|
|
routeUrls = append(routeUrls, u)
|
|
}
|
|
return routeUrls
|
|
}
|
|
|
|
// This will merge the flag routes and override anything that was present.
|
|
func mergeRoutes(opts, flagOpts *Options) {
|
|
routeUrls := RoutesFromStr(flagOpts.RoutesStr)
|
|
if routeUrls == nil {
|
|
return
|
|
}
|
|
opts.Routes = routeUrls
|
|
opts.RoutesStr = flagOpts.RoutesStr
|
|
}
|
|
|
|
func RemoveSelfReference(clusterPort int, routes []*url.URL) ([]*url.URL, error) {
|
|
var cleanRoutes []*url.URL
|
|
cport := strconv.Itoa(clusterPort)
|
|
|
|
selfIPs := getInterfaceIPs()
|
|
for _, r := range routes {
|
|
host, port, err := net.SplitHostPort(r.Host)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if cport == port && isIpInList(selfIPs, getUrlIp(host)) {
|
|
Noticef("Self referencing IP found: ", r)
|
|
continue
|
|
}
|
|
cleanRoutes = append(cleanRoutes, r)
|
|
}
|
|
|
|
return cleanRoutes, nil
|
|
}
|
|
|
|
func isIpInList(list1 []net.IP, list2 []net.IP) bool {
|
|
for _, ip1 := range list1 {
|
|
for _, ip2 := range list2 {
|
|
if ip1.Equal(ip2) {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func getUrlIp(ipStr string) []net.IP {
|
|
ipList := []net.IP{}
|
|
|
|
ip := net.ParseIP(ipStr)
|
|
if ip != nil {
|
|
ipList = append(ipList, ip)
|
|
return ipList
|
|
}
|
|
|
|
hostAddr, err := net.LookupHost(ipStr)
|
|
if err != nil {
|
|
Errorf("Error looking up host with route hostname: %v", err)
|
|
return ipList
|
|
}
|
|
for _, addr := range hostAddr {
|
|
ip = net.ParseIP(addr)
|
|
if ip != nil {
|
|
ipList = append(ipList, ip)
|
|
}
|
|
}
|
|
return ipList
|
|
}
|
|
|
|
func getInterfaceIPs() []net.IP {
|
|
var localIPs []net.IP
|
|
|
|
interfaceAddr, err := net.InterfaceAddrs()
|
|
if err != nil {
|
|
Errorf("Error getting self referencing address: %v", err)
|
|
return localIPs
|
|
}
|
|
|
|
for i := 0; i < len(interfaceAddr); i++ {
|
|
interfaceIP, _, _ := net.ParseCIDR(interfaceAddr[i].String())
|
|
if net.ParseIP(interfaceIP.String()) != nil {
|
|
localIPs = append(localIPs, interfaceIP)
|
|
} else {
|
|
Errorf("Error parsing self referencing address: %v", err)
|
|
}
|
|
}
|
|
return localIPs
|
|
}
|
|
|
|
func processOptions(opts *Options) {
|
|
// Setup non-standard Go defaults
|
|
if opts.Host == "" {
|
|
opts.Host = DEFAULT_HOST
|
|
}
|
|
if opts.Port == 0 {
|
|
opts.Port = DEFAULT_PORT
|
|
} else if opts.Port == RANDOM_PORT {
|
|
// Choose randomly inside of net.Listen
|
|
opts.Port = 0
|
|
}
|
|
if opts.MaxConn == 0 {
|
|
opts.MaxConn = DEFAULT_MAX_CONNECTIONS
|
|
}
|
|
if opts.PingInterval == 0 {
|
|
opts.PingInterval = DEFAULT_PING_INTERVAL
|
|
}
|
|
if opts.MaxPingsOut == 0 {
|
|
opts.MaxPingsOut = DEFAULT_PING_MAX_OUT
|
|
}
|
|
if opts.SslTimeout == 0 {
|
|
opts.SslTimeout = float64(SSL_TIMEOUT) / float64(time.Second)
|
|
}
|
|
if opts.AuthTimeout == 0 {
|
|
opts.AuthTimeout = float64(AUTH_TIMEOUT) / float64(time.Second)
|
|
}
|
|
if opts.ClusterAuthTimeout == 0 {
|
|
opts.ClusterAuthTimeout = float64(AUTH_TIMEOUT) / float64(time.Second)
|
|
}
|
|
if opts.MaxControlLine == 0 {
|
|
opts.MaxControlLine = MAX_CONTROL_LINE_SIZE
|
|
}
|
|
if opts.MaxPayload == 0 {
|
|
opts.MaxPayload = MAX_PAYLOAD_SIZE
|
|
}
|
|
}
|