func (r *roleService) RefreshRoleTokenCache(ctx context.Context) <-chan error {
	glg.Info("RefreshRoleTokenCache started")
	echan := make(chan error, r.domainRoleCache.Len()*(r.errRetryMaxCount+1))
	go func() {
		defer close(echan)
		r.domainRoleCache.Foreach(ctx, func(key string, val interface{}, exp int64) bool {
			domain, role, principal := decode(key)
			cd := val.(*cacheData)
			for err := range r.updateRoleTokenWithRetry(ctx, domain, role, principal, cd.minExpiry, cd.maxExpiry) {
				echan <- err
			}
			return true
		})
	}()
	return echan
}
RefreshRoleTokenCache returns the error channel when it is updated.
package service
import (
	"bytes"
	"context"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync/atomic"
	"time"
	"github.com/kpango/fastime"
	"github.com/kpango/gache"
	"github.com/kpango/glg"
	"github.com/kpango/ntokend"
	"github.com/pkg/errors"
	"github.com/yahoojapan/athenz-client-sidecar/v2/config"
	"golang.org/x/sync/singleflight"
)
type RoleService interface {
	StartRoleUpdater(context.Context) <-chan error
	RefreshRoleTokenCache(ctx context.Context) <-chan error
	GetRoleProvider() RoleProvider
}
type roleService struct {
	cfg                   config.RoleToken
	token                 ntokend.TokenProvider
	athenzURL             string
	athenzPrincipleHeader string
	domainRoleCache       gache.Gache
	group                 singleflight.Group
	expiry                time.Duration
	httpClient            atomic.Value
	rootCAs               *x509.CertPool
	certPath              string
	certKeyPath           string
	refreshPeriod    time.Duration
	errRetryMaxCount int
	errRetryInterval time.Duration
}
type cacheData struct {
	token             *RoleToken
	domain            string
	role              string
	proxyForPrincipal string
	minExpiry         int64
	maxExpiry         int64
}
type RoleToken struct {
	Token      string `json:"token"`
	ExpiryTime int64  `json:"expiryTime"`
}
type RoleProvider func(ctx context.Context, domain string, role string, proxyForPrincipal string, minExpiry int64, maxExpiry int64) (*RoleToken, error)
var (
	ErrRoleTokenRequestFailed = errors.New("Failed to fetch RoleToken")
	ErrInvalidSetting = errors.New("Invalid config")
	ErrDisabled = errors.New("Disabled")
	ErrNoCredentials = errors.New("No credentials")
)
const (
	defaultExpiry = time.Duration(0)
	defaultRefreshPeriod = time.Minute * 30
	defaultErrRetryMaxCount = 5
	defaultErrRetryInterval = time.Second * 5
	cacheKeySeparator = ";"
	roleSeparator = ","
	cachePurgePeriod = time.Minute
)
func NewRoleService(cfg config.RoleToken, token ntokend.TokenProvider) (RoleService, error) {
	var (
		err              error
		exp              = defaultExpiry
		refreshPeriod    = defaultRefreshPeriod
		errRetryInterval = defaultErrRetryInterval
	)
	if !cfg.Enable {
		return nil, ErrDisabled
	}
	if cfg.Expiry != "" {
		if exp, err = time.ParseDuration(cfg.Expiry); err != nil {
			return nil, errors.Wrap(ErrInvalidSetting, "Expiry: "+err.Error())
		}
	}
	if cfg.RefreshPeriod != "" {
		if refreshPeriod, err = time.ParseDuration(cfg.RefreshPeriod); err != nil {
			return nil, errors.Wrap(ErrInvalidSetting, "RefreshPeriod: "+err.Error())
		}
	}
	if cfg.Retry.Delay != "" {
		if errRetryInterval, err = time.ParseDuration(cfg.Retry.Delay); err != nil {
			return nil, errors.Wrap(ErrInvalidSetting, "ErrRetryInterval: "+err.Error())
		}
	}
	if exp != 0 && refreshPeriod > exp {
		return nil, errors.Wrap(ErrInvalidSetting, "refresh period > token expiry time")
	}
	errRetryMaxCount := defaultErrRetryMaxCount
	if cfg.Retry.Attempts > 0 {
		errRetryMaxCount = cfg.Retry.Attempts
	} else if cfg.Retry.Attempts != 0 {
		return nil, errors.Wrap(ErrInvalidSetting, "ErrRetryMaxCount < 0")
	}
	if token == nil && cfg.CertPath == "" {
		return nil, errors.Wrap(ErrInvalidSetting, "Neither NToken nor client certificate is set.")
	}
	var cp *x509.CertPool
	if cfg.AthenzCAPath != "" {
		var err error
		caPath := config.GetActualValue(cfg.AthenzCAPath)
		_, err = os.Stat(caPath)
		if os.IsNotExist(err) {
			return nil, errors.Wrap(ErrInvalidSetting, "Athenz CA not exist")
		}
		cp, err = NewX509CertPool(caPath)
		if err != nil {
			return nil, errors.Wrap(ErrInvalidSetting, err.Error())
		}
	}
	certPath := cfg.CertPath
	certKeyPath := cfg.CertKeyPath
	if token != nil {
		certPath = ""
		certKeyPath = ""
	}
	tlsConfig, err := NewTLSClientConfig(cp, certPath, certKeyPath)
	if err != nil {
		return nil, errors.Wrap(ErrInvalidSetting, err.Error())
	}
	var httpClient atomic.Value
	httpClient.Store(&http.Client{
		Transport: &http.Transport{
			TLSClientConfig: tlsConfig,
		},
	})
	return &roleService{
		cfg:                   cfg,
		token:                 token,
		athenzURL:             cfg.AthenzURL,
		athenzPrincipleHeader: cfg.PrincipalAuthHeader,
		domainRoleCache:       gache.New(),
		expiry:                exp,
		httpClient:            httpClient,
		rootCAs:               cp,
		certPath:              certPath,
		certKeyPath:           certKeyPath,
		refreshPeriod:         refreshPeriod,
		errRetryMaxCount:      errRetryMaxCount,
		errRetryInterval:      errRetryInterval,
	}, nil
}
func (r *roleService) StartRoleUpdater(ctx context.Context) <-chan error {
	glg.Info("Starting role token updater")
	ech := make(chan error, 100)
	go func() {
		defer close(ech)
		ticker := time.NewTicker(r.refreshPeriod)
		for {
			select {
			case <-ctx.Done():
				glg.Info("Stopping role token updater")
				ticker.Stop()
				ech <- ctx.Err()
				return
			case <-ticker.C:
				for err := range r.RefreshRoleTokenCache(ctx) {
					ech <- errors.Wrap(err, "error update role token")
				}
			}
		}
	}()
	r.domainRoleCache.StartExpired(ctx, cachePurgePeriod)
	r.domainRoleCache.EnableExpiredHook().SetExpiredHook(func(ctx context.Context, k string) {
		glg.Warnf("the following cache is expired, key: %v", k)
	})
	return ech
}
func (r *roleService) GetRoleProvider() RoleProvider {
	return r.getRoleToken
}
func (r *roleService) getRoleToken(ctx context.Context, domain, role, proxyForPrincipal string, minExpiry, maxExpiry int64) (*RoleToken, error) {
	tok, ok := r.getCache(domain, role, proxyForPrincipal)
	if !ok {
		return r.updateRoleToken(ctx, domain, role, proxyForPrincipal, minExpiry, maxExpiry)
	}
	return tok, nil
}