}

	switch {
	case p[0].Key != nil:
		key := p[0].Key.keyString()

		switch kvs := v.(type) {
		case jstream.KVS:
			for _, kv := range kvs {
				if kv.Key == key {
					return jsonpathEval(p[1:], kv.Value)
				}
			}
			// Key not found - return nil result
			return Missing{}, false, nil
		case simdjson.Object:
			elem := kvs.FindKey(key, nil)
			if elem == nil {
				// Key not found - return nil result

		return cfg, err
	}

	hstsIncludeSubdomains := env.Get(EnvBrowserHSTSIncludeSubdomains, kvs.GetWithDefault(browserHSTSIncludeSubdomains, DefaultKVS)) == config.EnableOn
	hstsPreload := env.Get(EnvBrowserHSTSPreload, kvs.Get(browserHSTSPreload)) == config.EnableOn

	hstsSeconds, err := strconv.Atoi(env.Get(EnvBrowserHSTSSeconds, kvs.GetWithDefault(browserHSTSSeconds, DefaultKVS)))

	}
	return s.String()
}

// KVS - is a shorthand for some wrapper functions
// to operate on list of key values.
type KVS []KV

// Empty - return if kv is empty
func (kvs KVS) Empty() bool {
	return len(kvs) == 0
}

// Clone - returns a copy of the KVS
func (kvs KVS) Clone() KVS {
	return append(make(KVS, 0, len(kvs)), kvs...)
}

	// has not been changed from its "default" value, if it has been changed honor new settings.
	if kvs.GetWithDefault(Speed, DefaultKVS) == "default" {
		if kvs.Get(Delay) != "" && kvs.Get(MaxWait) != "" {
			if err = lookupDeprecatedScannerConfig(kvs, &cfg); err != nil {
				return cfg, err
			}
		}
	}

	switch speed := env.Get(EnvSpeed, kvs.GetWithDefault(Speed, DefaultKVS)); speed {
	case "fastest":

		return nil, errDataSource(errors.New("path not supported"))
	}
	switch rec := rawVal.(type) {
	case jstream.KVS:
		txedRec, _, err := jsonpathEval(e.selectAST.From.Table.PathExpr[1:], rec)
		if err != nil {
			return nil, err
		}

		var kvs jstream.KVS
		switch v := txedRec.(type) {
		case jstream.KVS:
			kvs = v

		case []interface{}:
			recs := make([]*Record, len(v))
			for i, val := range v {

func LookupConfig(kvs config.KVS) (cfg Config, err error) {
	cfg = Config{
		TransitionWorkers: 100,
		ExpirationWorkers: 100,
	}

	if err = config.CheckValidKeys(config.ILMSubSys, kvs, DefaultKVS); err != nil {
		return cfg, err
	}

	tw, err := strconv.Atoi(env.Get(EnvILMTransitionWorkers, kvs.GetWithDefault(transitionWorkers, DefaultKVS)))
	if err != nil {
		return cfg, err

// LookupConfig - lookup config and override with valid environment settings if any.
func LookupConfig(kvs config.KVS) (cfg Config, err error) {
	cfg = Config{
		MaxTimeout: 30 * time.Second,
	}
	if err = config.CheckValidKeys(config.DriveSubSys, kvs, DefaultKVS); err != nil {
		return cfg, err
	}
	// if not set. Get default value from environment
	d := kvs.GetWithDefault(MaxTimeout, DefaultKVS)
	if d == "" {

func LookupConfig(kvs config.KVS) (cfg Config, err error) {
	deprecatedKeys := []string{
		apiReadyDeadline,
		"extend_list_cache_life",
		apiReplicationWorkers,
		apiReplicationFailedWorkers,
		"expiry_workers",
	}

	disableODirect := env.Get(EnvAPIDisableODirect, kvs.Get(apiDisableODirect)) == config.EnableOn
	enableODirect := env.Get(EnvAPIODirect, kvs.Get(apiODirect)) == config.EnableOn

func SetNotifyKafka(s config.Config, name string, cfg target.KafkaArgs) error {
	if !cfg.Enable {
		return nil
	}

	if err := cfg.Validate(); err != nil {
		return err
	}

	s[config.NotifyKafkaSubSys][name] = config.KVS{
		config.KV{
			Key:   config.Enable,
			Value: config.EnableOn,
		},
		config.KV{
			Key: target.KafkaBrokers,
			Value: func() string {
				var brokers []string
				for _, broker := range cfg.Brokers {

	}
	for k, v := range notify.DefaultNotificationKVS {
		kvs[k] = v
	}
	for k, v := range lambda.DefaultLambdaKVS {
		kvs[k] = v
	}
	if globalIsErasure {
		kvs[config.StorageClassSubSys] = storageclass.DefaultKVS
		kvs[config.HealSubSys] = heal.DefaultKVS
	}
	config.RegisterDefaultKVS(kvs)

	// Captures help for each sub-system
	helpSubSys := config.HelpKVS{
		config.HelpKV{