Can't reliably get PubSub - "no available client found"

[nicoroy2561]

I keep getting the "no available client found" error when calling pubsub, err = client.NewPubSub() on client = cache.NewEmbeddedClient(). It seems to seldomly work - I managed to get it to work like 1/10 times.

Details

22:36:17.574 [INFO] Olric 0.5.4 on linux/amd64 go1.22.0
 
23:36:17.574 [INFO] Join completed. Synced with 1 initial nodes
 
23:36:17.575 [INFO] Memberlist bindAddr: 10.244.0.88, bindPort: 3322
 
23:36:17.575 [INFO] Cluster coordinator: 10.244.0.38:3320
 
23:36:17.575 [INFO] Node name in the cluster: 10.244.0.88:3320
 
23:36:17.575 [INFO] Olric bindAddr: 10.244.0.88, bindPort: 3320
 
23:36:17.575 [INFO] Replication count is 1
 
23:36:17.575 [INFO] Node joined: 10.244.0.38:3320
 
23:36:17.587 [INFO] Routing table has been pushed by 10.244.0.38:3320
 
23:36:17.588 pubsub error no available client found

Code I'm using is the one from the base example with some custom kubernetes discovery setup:

func Init(discovery *k8s.K8sDiscovery) bool {
	discoveryAdapter, err := newK8sAdapter(discovery)
	if err != nil {
		log.Fatalln("adapter error:", err)
	}

	// create a new Olric configuration
	cfg := config.New("lan") // default configuration
	cfg.ReplicationMode = config.SyncReplicationMode
	cfg.ServiceDiscovery = map[string]interface{}{"plugin": discoveryAdapter} //https://gist.github.com/derekperkins/8f5232d897ccf5a20691a1556574cfce
	cfg.DMaps.EvictionPolicy = config.LRUEviction                             //
	cfg.DMaps.MaxInuse = 50_000_000                                           // 50 MB

	cfg.LogLevel = "INFO"
	cfg.LogVerbosity = 3 // at least 2 is recommended; https://github.com/buraksezer/olric/blob/master/config/config.go#L102

	// this wait group is used to block the main goroutine until the embedded client is ready
	wg := sync.WaitGroup{}
	wg.Add(1)
	cfg.Started = func() { wg.Done() }

	// create the actual Olric instance
	cache, err = olric.New(cfg)
	if err != nil {
		log.Println("error creating cache from config:", err)
		return false
	}

	// start the instance, which triggers the k8s service discovery and forms the cluster
	go func() {
		if err := cache.Start(); err != nil {
			panic("error starting:" + err.Error())
		}
	}()

	// wait for the cluster to be ready before continuing
	wg.Wait()

	client = cache.NewEmbeddedClient()
	if client == nil {
		log.Fatalln("client nil")
	}

	pubsub, err = client.NewPubSub()
	if err != nil {
		log.Fatalln("pubsub error", err)
	}
	return true
}

Any suggestions on what I might be doing wrong here?

[xadhatter]

I ran into this as well. If you look closely at the PubSub example in the README, it uses a ClusterClient, not EmbeddedClient.

  c, err := olric.NewClusterClient([]string{"localhost:3320"})
  if err != nil {
    log.Fatalf("olric.NewClusterClient returned an error: %v", err)
  }

Reading through the code a bit, one of the differences between the two is that ClusterClient initializes clients for the cluster using the provided bootstrap address when it starts up. Since the EmbeddedClient does not do this you get the "no available client found" error.

It is unclear to me if this is expected behavior or not, as we both made the same mistake. There are two solutions that worked for me, not sure if one is more "correct". First, use a ClusterClient. Second, provide the local address as the ToAddress of the PubSub when you create it.

pubsub, err := cl.NewPubSub(olric.ToAddress("localhost:3320"))

This basically forces the initialization of the local client and uses it for the PubSub.